Merge branch 'main' into release/current

Author: SamuelHassine

.gitignore

@@ -133,3 +133,6 @@ dmypy.json
 
 # Cython debug symbols
 cython_debug/
+
+# testing
+test.py

pyobas/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-__version__ = "1.18.12"
+__version__ = "1.18.16"
 
 from pyobas._version import (  # noqa: F401
     __author__,

pyobas/_version.py

@@ -3,4 +3,4 @@
 __email__ = "[email protected]"
 __license__ = "Apache 2.0"
 __title__ = "python-openbas"
-__version__ = "1.18.12"
+__version__ = "1.18.16"

pyobas/apis/endpoint.py

@@ -2,6 +2,7 @@
 
 from pyobas import exceptions as exc
 from pyobas.base import RESTManager, RESTObject
+from pyobas.utils import RequiredOptional
 
 
 class Endpoint(RESTObject):
@@ -11,9 +12,27 @@ class Endpoint(RESTObject):
 class EndpointManager(RESTManager):
     _path = "/endpoints"
     _obj_cls = Endpoint
+    _create_attrs = RequiredOptional(
+        required=(
+            "endpoint_hostname",
+            "endpoint_platform",
+            "endpoint_arch",
+        ),
+        optional=(
+            "endpoint_mac_addresses",
+            "endpoint_ips",
+            "asset_external_reference",
+        ),
+    )
 
     @exc.on_http_error(exc.OpenBASUpdateError)
     def get(self, asset_id: str, **kwargs: Any) -> Dict[str, Any]:
         path = f"{self.path}/" + asset_id
         result = self.openbas.http_get(path, **kwargs)
         return result
+
+    @exc.on_http_error(exc.OpenBASUpdateError)
+    def upsert(self, endpoint: Dict[str, Any], **kwargs: Any) -> Dict[str, Any]:
+        path = f"{self.path}/agentless/upsert"
+        result = self.openbas.http_post(path, post_data=endpoint, **kwargs)
+        return result

pyobas/client.py

@@ -197,7 +197,6 @@ def http_request(
                 stream=streamed,
                 **opts,
             )
-
             self._check_redirects(result.response)
 
             if 200 <= result.status_code < 300:
@@ -225,25 +224,8 @@ def http_request(
                     error_json = result.json()
                     # Common fields
                     if isinstance(error_json, dict):
-                        # First priority: look for a 'message' field (most specific)
-                        if "message" in error_json:
-                            error_message = error_json.get("message")
-                        elif "execution_message" in error_json:
-                            error_message = error_json.get("execution_message")
-                        elif "error" in error_json:
-                            err = error_json.get("error")
-                            if isinstance(err, dict) and "message" in err:
-                                error_message = err.get("message")
-                            elif err and err not in [
-                                "Internal Server Error",
-                                "Bad Request",
-                                "Not Found",
-                                "Unauthorized",
-                                "Forbidden",
-                            ]:
-                                # Only use 'error' field if it's not a generic HTTP status
-                                error_message = str(err)
-                        elif "errors" in error_json:
+                        # Check for nested validation errors first (more specific)
+                        if "errors" in error_json:
                             errs = error_json.get("errors")
                             if isinstance(errs, list) and errs:
                                 # Join any messages in the list
@@ -254,6 +236,52 @@ def http_request(
                                     else:
                                         messages.append(str(item))
                                 error_message = "; ".join(messages)
+                            elif isinstance(errs, dict):
+                                # Handle nested validation errors from OpenBAS
+                                if "children" in errs:
+                                    # This is a validation error structure
+                                    validation_errors = []
+                                    children = errs.get("children", {})
+                                    for field, field_errors in children.items():
+                                        if (
+                                            isinstance(field_errors, dict)
+                                            and "errors" in field_errors
+                                        ):
+                                            field_error_list = field_errors.get(
+                                                "errors", []
+                                            )
+                                            if field_error_list:
+                                                for err_msg in field_error_list:
+                                                    validation_errors.append(
+                                                        f"{field}: {err_msg}"
+                                                    )
+                                    if validation_errors:
+                                        base_msg = error_json.get(
+                                            "message", "Validation Failed"
+                                        )
+                                        error_message = f"{base_msg}: {'; '.join(validation_errors)}"
+                                elif isinstance(errs, str):
+                                    error_message = errs
+
+                        # If no error message from errors field, check other fields
+                        if not error_message:
+                            if "message" in error_json:
+                                error_message = error_json.get("message")
+                            elif "execution_message" in error_json:
+                                error_message = error_json.get("execution_message")
+                            elif "error" in error_json:
+                                err = error_json.get("error")
+                                if isinstance(err, dict) and "message" in err:
+                                    error_message = err.get("message")
+                                elif err and err not in [
+                                    "Internal Server Error",
+                                    "Bad Request",
+                                    "Not Found",
+                                    "Unauthorized",
+                                    "Forbidden",
+                                ]:
+                                    # Only use 'error' field if it's not a generic HTTP status
+                                    error_message = str(err)
                     elif isinstance(error_json, str):
                         error_message = error_json
                     # Fallback to serialized json if we still have nothing

pyobas/exceptions.py

@@ -39,6 +39,7 @@ def __str__(self) -> str:
             "Service Unavailable",
             "Gateway Timeout",
             "Unknown error",
+            "Validation Failed",
         )
 
         # Only try to extract from response body if message is truly generic
@@ -78,6 +79,37 @@ def __str__(self) -> str:
                                 else:
                                     parts.append(str(item))
                             extracted_msg = "; ".join(parts)
+                        elif isinstance(errs, dict):
+                            # Handle nested validation errors structure
+                            if "children" in errs:
+                                validation_errors = []
+                                children = errs.get("children", {})
+                                for field, field_errors in children.items():
+                                    if (
+                                        isinstance(field_errors, dict)
+                                        and "errors" in field_errors
+                                    ):
+                                        field_error_list = field_errors.get(
+                                            "errors", []
+                                        )
+                                        if field_error_list:
+                                            for err_msg in field_error_list:
+                                                validation_errors.append(
+                                                    f"{field}: {err_msg}"
+                                                )
+                                if validation_errors:
+                                    base_msg = data.get("message", "Validation Failed")
+                                    extracted_msg = (
+                                        f"{base_msg}: {'; '.join(validation_errors)}"
+                                    )
+                            else:
+                                # Try to get any string representation
+                                parts = []
+                                for key, value in errs.items():
+                                    if value:
+                                        parts.append(f"{key}: {value}")
+                                if parts:
+                                    extracted_msg = "; ".join(parts)
                         elif isinstance(errs, str):
                             extracted_msg = errs