Merge pull request #1 from crowdsecurity/$main-032f7c1

Update python SDK version: 1.0.0

Author: he2ss

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 Crowdsec
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,21 @@
+# crowdsec_tracker_api
+
+**crowdsec_tracker_api** is a Python SDK for the [CrowdSec Tracker API](https://docs.crowdsec.net/u/service_api/intro/).
+This library enables you to manage CrowdSec Live Exploit Tracker resources such as CVEs data, IPs from a specific CVE and manage integrations.
+
+## Installation
+
+```bash
+pip install crowdsec_tracker_api
+```
+
+## Usage
+
+You can follow this [documentation](https://docs.crowdsec.net/u/service_api/quickstart/blocklists) to see the basic usage of the SDK.
+
+## Documentation
+You can access the full usage documentation [here](https://github.com/crowdsecurity/crowdsec-tracker-api-sdk-python/tree/main/doc).
+
+## Contributing
+
+This SDK is auto-generated from the API specification. You can open an issue on this repository if you find a bug or if you want to add a feature.

crowdsec_tracker_api/__init__.py

@@ -0,0 +1,56 @@
+from enum import Enum
+from .models import *
+from .base_model import Page
+from .services.integrations import Integrations
+from .services.cves import Cves
+from .http_client import ApiKeyAuth
+
+class Server(Enum):
+    production_server = 'https://admin.api.crowdsec.net/v1'
+
+__all__ = [
+    'Integrations',
+    'Cves',
+    'ApiKeyCredentials',
+    'BasicAuthCredentials',
+    'BlocklistSubscription',
+    'HTTPValidationError',
+    'IntegrationCreateRequest',
+    'IntegrationCreateResponse',
+    'IntegrationGetResponse',
+    'IntegrationGetResponsePage',
+    'IntegrationType',
+    'IntegrationUpdateRequest',
+    'IntegrationUpdateResponse',
+    'Links',
+    'OutputFormat',
+    'Stats',
+    'ValidationError',
+    'AffectedComponent',
+    'AllowlistSubscription',
+    'AttackDetail',
+    'Behavior',
+    'CVEEvent',
+    'CVEResponseBase',
+    'CVEsubscription',
+    'CWE',
+    'Classification',
+    'Classifications',
+    'EntityType',
+    'GetCVEIPsResponsePage',
+    'GetCVEResponse',
+    'GetCVESubscribedIntegrationsResponsePage',
+    'GetCVEsResponsePage',
+    'History',
+    'IPItem',
+    'IntegrationResponse',
+    'Location',
+    'MitreTechnique',
+    'Reference',
+    'ScoreBreakdown',
+    'Scores',
+    'SubscribeCVEIntegrationRequest',
+    'ApiKeyAuth',
+    'Server',
+    'Page'
+]

crowdsec_tracker_api/__pycache__/base_model.cpython-311.pyc

@@ -0,0 +1,58 @@
+from urllib.parse import urlparse
+from pydantic import BaseModel, ConfigDict, PrivateAttr, RootModel
+from typing import Generic, Sequence, Optional, TypeVar, Any
+from httpx import Auth
+from .http_client import HttpClient
+
+
+class BaseModelSdk(BaseModel):
+    model_config = ConfigDict(
+        extra="ignore",
+    )
+    _client: Optional["Service"] = PrivateAttr(default=None)
+
+    def __init__(self, /, _client: "Service" = None, **data):
+        super().__init__(**data)
+        self._client = _client
+
+    def next(self, client: "Service" = None) -> Optional["BaseModelSdk"]:
+        return (client if client is not None else self._client).next_page(self)
+
+
+class RootModelSdk(RootModel):
+    def __getattr__(self, item: str) -> Any:
+        return getattr(self.root, item)
+
+
+T = TypeVar("T")
+
+
+class Page(BaseModelSdk, Generic[T]):
+    items: Sequence[T]
+    total: Optional[int]
+    page: Optional[int]
+    size: Optional[int]
+    pages: Optional[int] = None
+    links: Optional[dict] = None
+
+
+class Service:
+    def __init__(self, base_url: str, auth: Auth) -> None:
+        self.http_client = HttpClient(base_url=base_url, auth=auth)
+
+    def next_page(self, page: BaseModelSdk) -> Optional[BaseModelSdk]:
+        if not hasattr(page, "links") or not page.links:
+            raise ValueError(
+                "No links found in the response, this is not a paginated response."
+            )
+        if page.links.next:
+            # links are relative to host not to full base url. We need to pass a full formatted url here
+            parsed_url = urlparse(self.http_client.base_url)
+            response = self.http_client.get(
+                f"{parsed_url.scheme}://{parsed_url.netloc}{page.links.next}",
+                path_params=None,
+                params=None,
+                headers=None,
+            )
+            return page.__class__(_client=self, **response.json())
+        return None

crowdsec_tracker_api/__pycache__/http_client.cpython-311.pyc

@@ -0,0 +1,153 @@
+from urllib.parse import quote, urlparse
+
+from botocore.auth import SigV4Auth
+from botocore.awsrequest import AWSRequest
+import botocore.session
+
+import httpx
+
+
+class AWSSignV4Auth(httpx.Auth):
+    def __init__(self, aws_region="eu-west-1") -> None:
+        self.aws_region = aws_region
+
+    def auth_flow(self, request):
+        request = self.sign_request(request)
+        yield request
+
+    def sign_request(self, request: httpx.Request) -> httpx.Request:
+        """Signs an httpx request with AWS Signature Version 4."""
+
+        session = botocore.session.get_session()
+        credentials = session.get_credentials()
+        aws_request = AWSRequest(
+            method=request.method.upper(), url=str(request.url), data=request.content
+        )
+        region = self.aws_region
+        service = "execute-api"
+
+        # Sign the request
+        SigV4Auth(credentials, service, region).add_auth(aws_request)
+
+        # Update the httpx request headers with the signed headers
+        request.headers.update(dict(aws_request.headers))
+
+        return request
+
+
+class ApiKeyAuth(httpx.Auth):
+    def __init__(self, api_key: str) -> None:
+        self.api_key = api_key
+
+    def auth_flow(self, request):
+        request.headers["x-api-key"] = self.api_key
+        yield request
+
+
+class HttpClient:
+    def __init__(self, base_url: str, auth: httpx.Auth, aws_region="eu-west-1") -> None:
+        self.aws_region = aws_region
+        self.base_url = base_url
+        self.auth = auth
+        self.client = httpx.Client()
+        self.timeout = 30
+
+    def _replace_path_params(self, url: str, path_params: dict):
+        if path_params:
+            for param, value in path_params.items():
+                if not value:
+                    raise ValueError(
+                        f"Parameter {param} is required, cannot be empty or blank."
+                    )
+                url = url.replace(f"{{{param}}}", quote(str(value)))
+        return url
+
+    def _normalize_url(self, url: str):
+        self.base_url = self.base_url.rstrip("/")
+        parsed_url = urlparse(url)
+        if not parsed_url.netloc:
+            return f"{self.base_url}{url}"
+        return url
+
+    def get(
+        self,
+        url: str,
+        path_params: dict = None,
+        params: dict = None,
+        headers: dict = None,
+    ):
+        url = self._replace_path_params(
+            url=self._normalize_url(url), path_params=path_params
+        )
+        response = self.client.get(
+            url=url,
+            params=params,
+            headers=headers,
+            auth=self.auth,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        return response
+
+    def post(
+        self, url: str, path_params: dict, params: dict, headers: dict, json: dict
+    ):
+        url = self._replace_path_params(
+            url=self._normalize_url(url), path_params=path_params
+        )
+        response = self.client.post(
+            url=url,
+            params=params,
+            headers=headers,
+            json=json,
+            auth=self.auth,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        return response
+
+    def put(self, url: str, path_params: dict, params: dict, headers: dict, json: dict):
+        url = self._replace_path_params(
+            url=self._normalize_url(url), path_params=path_params
+        )
+        response = self.client.put(
+            url=url,
+            params=params,
+            headers=headers,
+            json=json,
+            auth=self.auth,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        return response
+
+    def patch(
+        self, url: str, path_params: dict, params: dict, headers: dict, json: dict
+    ):
+        url = self._replace_path_params(
+            url=self._normalize_url(url), path_params=path_params
+        )
+        response = self.client.patch(
+            url=url,
+            params=params,
+            headers=headers,
+            json=json,
+            auth=self.auth,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        return response
+
+    def delete(self, url: str, path_params: dict, params: dict, headers: dict):
+        url = self._replace_path_params(
+            url=self._normalize_url(url), path_params=path_params
+        )
+        response = self.client.delete(
+            url=url,
+            params=params,
+            headers=headers,
+            auth=self.auth,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        return response