Merge pull request #50 from GitGuardian/alacombe/implement-honeytoken-creation

implement `create_honeytoken` method

Author: agateau-gg

changelog.d/20230425_100744_antonin.lacombe_implement_honeytoken_creation.md

@@ -0,0 +1,4 @@
+
+### Added
+
+- Add `GGClient.create_honeytoken()` method.

pygitguardian/client.py

@@ -25,6 +25,7 @@
     Detail,
     Document,
     HealthCheckResponse,
+    HoneytokenResponse,
     MultiScanResult,
     QuotaResponse,
     ScanResult,
@@ -80,6 +81,17 @@ def is_ok(resp: Response) -> bool:
     )
 
 
+def is_create_ok(resp: Response) -> bool:
+    """
+    is_create_ok returns True if the API returns code 201
+    and the content type is JSON.
+    """
+    return (
+        resp.headers["content-type"] == "application/json"
+        and resp.status_code == codes.created
+    )
+
+
 def _create_tar(root_path: Path, filenames: List[str]) -> bytes:
     """
     :param root_path: the root_path from which the tar is created
@@ -390,6 +402,43 @@ def quota_overview(
 
         return obj
 
+    def create_honeytoken(
+        self,
+        name: Optional[str],
+        type_: str,
+        description: Optional[str],
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Detail, HoneytokenResponse]:
+        """
+        Create a honeytoken via the /honeytokens endpoint of the API
+
+        :param name: the honeytoken name
+        :param type_: the honeytoken type
+        :param description: the honeytoken description
+        :param extra_headers: additional headers to add to the request
+        :return: Detail or Honeytoken response and status code
+        """
+        try:
+            resp = self.post(
+                endpoint="honeytokens",
+                extra_headers=extra_headers,
+                data={
+                    "name": name,
+                    "type": type_,
+                    "description": description,
+                },
+            )
+        except requests.exceptions.ReadTimeout:
+            result = Detail("The request timed out.")
+            result.status_code = 504
+        else:
+            if is_create_ok(resp):
+                result = HoneytokenResponse.SCHEMA.load(resp.json())
+            else:
+                result = load_detail(resp)
+            result.status_code = resp.status_code
+        return result
+
     # For IaC Scans
     def iac_directory_scan(
         self,

pygitguardian/models.py

@@ -1,4 +1,4 @@
-from datetime import date
+from datetime import date, datetime
 from typing import Any, ClassVar, Dict, List, Optional, cast
 
 from marshmallow import (
@@ -485,6 +485,99 @@ def __repr__(self) -> str:
         return f"content:{repr(self.content)}"
 
 
+class HoneytokenResponseSchema(BaseSchema):
+    id = fields.Int()
+    name = fields.String()
+    description = fields.String(allow_none=True)
+    created_at = fields.AwareDateTime()
+    status = fields.String()
+    triggered_at = fields.AwareDateTime(allow_none=True)
+    revoked_at = fields.AwareDateTime(allow_none=True)
+    open_events_count = fields.Int(allow_none=True)
+    type_ = fields.String(data_key="type")
+    creator_id = fields.Int(allow_none=True)
+    revoker_id = fields.Int(allow_none=True)
+    creator_api_token_id = fields.String(allow_none=True)
+    revoker_api_token_id = fields.String(allow_none=True)
+    token = fields.Mapping(key=fields.String(), value=fields.String())
+    tags = fields.List(fields.String())
+
+    @post_load
+    def make_honeytoken_response(
+        self, data: Dict[str, Any], **kwargs: Any
+    ) -> "HoneytokenResponse":
+        return HoneytokenResponse(**data)
+
+
+class HoneytokenResponse(Base):
+    """
+    honeytoken creation in the GitGuardian API.
+    Allows users to create and get a honeytoken.
+    Example:
+        {
+            "id": 141,
+            "name": "honeytoken A",
+            "description": "honeytoken used in the repository AA",
+            "created_at": "2019-08-22T14:15:22Z",
+            "status": "active",
+            "triggered_at": "2019-08-22T14:15:22Z",
+            "revoked_at": "2019-08-22T14:15:22Z",
+            "open_events_count": 122,
+            "type": "AWS",
+            "creator_id": 122,
+            "revoker_id": 122,
+            "creator_api_token_id": null,
+            "revoker_api_token_id": null,
+            "token": {
+                "access_token_id": "AAAA",
+                "secret_key": "BBB"
+            },
+        "tags": ["publicly_exposed"]
+        }
+    """
+
+    SCHEMA = HoneytokenResponseSchema()
+
+    def __init__(
+        self,
+        id: int,
+        name: str,
+        description: Optional[str],
+        created_at: datetime,
+        status: str,
+        triggered_at: Optional[datetime],
+        revoked_at: Optional[datetime],
+        open_events_count: Optional[int],
+        type_: str,
+        creator_id: Optional[int],
+        revoker_id: Optional[int],
+        creator_api_token_id: Optional[str],
+        revoker_api_token_id: Optional[str],
+        token: Dict[str, str],
+        tags: List[str],
+        **kwargs: Any,
+    ) -> None:
+        super().__init__()
+        self.id = id
+        self.name = name
+        self.description = description
+        self.created_at = created_at
+        self.status = status
+        self.triggered_at = triggered_at
+        self.revoked_at = revoked_at
+        self.open_events_count = open_events_count
+        self.type_ = type_
+        self.creator_id = creator_id
+        self.revoker_id = revoker_id
+        self.creator_api_token_id = creator_api_token_id
+        self.revoker_api_token_id = revoker_api_token_id
+        self.token = token
+        self.tags = tags
+
+    def __repr__(self) -> str:
+        return f"honeytoken:{self.id} {self.name}"
+
+
 class HealthCheckResponseSchema(BaseSchema):
     detail = fields.String(allow_none=False)
     status_code = fields.Int(allow_none=False)

tests/test_client.py

@@ -16,7 +16,13 @@
     DOCUMENT_SIZE_THRESHOLD_BYTES,
     MULTI_DOCUMENT_LIMIT,
 )
-from pygitguardian.models import Detail, MultiScanResult, QuotaResponse, ScanResult
+from pygitguardian.models import (
+    Detail,
+    HoneytokenResponse,
+    MultiScanResult,
+    QuotaResponse,
+    ScanResult,
+)
 
 from .conftest import my_vcr
 
@@ -646,3 +652,75 @@ def test_versions_from_headers(request_mock: Mock, client: GGClient, method):
     other_client = GGClient(api_key="")
     assert other_client.app_version is app_version_value
     assert other_client.secrets_engine_version is secrets_engine_version_value
+
+
+@patch("requests.Session.request")
+def test_create_honeytoken(
+    request_mock: Mock,
+    client: GGClient,
+):
+    """
+    GIVEN a ggclient
+    WHEN calling create_honeytoken with parameters
+    THEN the parameters are passed in the request and the returned honeytoken use the parameters
+    """
+    mock_response = Mock(spec=Response)
+    mock_response.headers = {"content-type": "application/json"}
+    mock_response.status_code = 201
+    mock_response.json.return_value = {
+        "id": 141,
+        "name": "honeytoken A",
+        "description": "honeytoken used in the repository AA",
+        "created_at": "2019-08-22T14:15:22Z",
+        "status": "active",
+        "triggered_at": "2019-08-22T14:15:22Z",
+        "revoked_at": None,
+        "open_events_count": 2,
+        "type": "AWS",
+        "creator_id": 122,
+        "revoker_id": None,
+        "creator_api_token_id": None,
+        "revoker_api_token_id": None,
+        "token": {
+            "access_token_id": "AAAA",
+            "secret_key": "BBB"
+        },
+        "tags": ["publicly_exposed"]
+    }
+
+    request_mock.return_value = mock_response
+
+    result = client.create_honeytoken(name="honeytoken A",
+                                      description="honeytoken used in the repository AA",
+                                      type_="AWS")
+
+    assert request_mock.called
+    assert isinstance(result, HoneytokenResponse)
+
+
+@patch("requests.Session.request")
+def test_create_honeytoken_error(
+    request_mock: Mock,
+    client: GGClient,
+):
+    """
+    GIVEN a ggclient
+    WHEN calling create_honeytoken with parameters without the right access
+    THEN I get a Detail objects containing the error detail
+    """
+    mock_response = Mock(spec=Response)
+    mock_response.headers = {"content-type": "application/json"}
+    mock_response.status_code = 400
+    mock_response.json.return_value = {
+        "detail": "Not authorized",
+    }
+
+    request_mock.return_value = mock_response
+
+    result = client.create_honeytoken(name="honeytoken A",
+                                      description="honeytoken used in the repository AA",
+                                      type_="AWS")
+
+    assert request_mock.called
+    assert isinstance(result, Detail)
+    result.status_code == 400

tests/test_models.py

@@ -8,6 +8,8 @@
     Document,
     DocumentSchema,
     HealthCheckResponseSchema,
+    HoneytokenResponse,
+    HoneytokenResponseSchema,
     Match,
     MatchSchema,
     MultiScanResult,
@@ -135,6 +137,30 @@ def test_document_handle_surrogates(self):
                 Detail,
                 {"detail": "Fail"},
             ),
+            (
+                HoneytokenResponseSchema,
+                HoneytokenResponse,
+                {
+                    "id": 141,
+                    "name": "honeytoken A",
+                    "description": "honeytoken used in the repository AA",
+                    "created_at": "2019-08-22T14:15:22Z",
+                    "status": "active",
+                    "triggered_at": "2019-08-22T14:15:22Z",
+                    "revoked_at": None,
+                    "open_events_count": 2,
+                    "type": "AWS",
+                    "creator_id": 122,
+                    "revoker_id": None,
+                    "creator_api_token_id": None,
+                    "revoker_api_token_id": None,
+                    "token": {
+                        "access_token_id": "AAAA",
+                        "secret_key": "BBB"
+                    },
+                    "tags": ["publicly_exposed"]
+                },
+            ),
         ],
     )
     def test_schema_loads(self, schema_klass, expected_klass, instance_data):