Merge pull request #23 from nodestream-proj/audit-log-actor-filter

Actor-based Filtering for GH AuditLogExtractor

Author: rreddy2

nodestream_github/audit.py

@@ -20,29 +20,38 @@
 class GithubAuditLogExtractor(Extractor):
     """
     Extracts audit logs from the GitHub REST API.
-    You can pass the enterprise_name, actions and lookback_period to the extractor
-    along with the regular GitHub parameters.
+    You can pass the enterprise_name, actions, actors, exclude_actors
+    and lookback_period to the extractor along with the regular
+    GitHub parameters.
 
     lookback_period can contain keys for days, months, and/or years as ints
-    actions can be found in the GitHub documentation
+    actions, and actors/exclude_actors can be found in the GitHub documentation
     https://docs.github.com/en/[email protected]/admin/monitoring-activity-in-your-enterprise/reviewing-audit-logs-for-your-enterprise/searching-the-audit-log-for-your-enterprise#search-based-on-the-action-performed
     """
 
     def __init__(
         self,
         enterprise_name: str,
         actions: list[str] | None = None,
+        actors: list[str] | None = None,
+        exclude_actors: list[str] | None = None,
         lookback_period: dict[str, int] | None = None,
         **github_client_kwargs: Any | None,
     ):
         self.enterprise_name = enterprise_name
         self.client = GithubRestApiClient(**github_client_kwargs)
         self.lookback_period = lookback_period
         self.actions = actions
+        self.actors = actors
+        self.exclude_actors = exclude_actors
 
     async def extract_records(self) -> AsyncGenerator[GithubAuditLog]:
         async for audit in self.client.fetch_enterprise_audit_log(
-            self.enterprise_name, self.actions, self.lookback_period
+            self.enterprise_name,
+            self.actions,
+            self.actors,
+            self.exclude_actors,
+            self.lookback_period,
         ):
             audit["timestamp"] = audit.pop("@timestamp")
             yield audit

nodestream_github/client/githubclient.py

@@ -74,6 +74,68 @@ def _fetch_problem(title: str, e: httpx.HTTPError):
             logger.warning("Problem fetching %s", title, exc_info=e, stacklevel=2)
 
 
+def validate_lookback_period(lookback_period: dict[str, int]) -> dict[str, int]:
+    """Sanitize the lookback period to only include valid keys."""
+
+    def validate_positive_int(value: int) -> int:
+        converted = int(value)
+        if converted <= 0:
+            negative_value_exception_msg = (
+                f"Lookback period values must be positive: {value}"
+            )
+            raise ValueError(negative_value_exception_msg)
+        return converted
+
+    try:
+        return {k: validate_positive_int(v) for k, v in lookback_period.items()}
+    except Exception as e:
+        exception_msg = "Formatting lookback period failed"
+        raise ValueError(exception_msg) from e
+
+
+def build_search_phrase(
+    actions: list[str],
+    actors: list[str],
+    exclude_actors: list[str],
+    lookback_period: dict[str, int],
+) -> str:
+    # adding action-based filtering
+    actions_phrase = ""
+    if actions:
+        actions_phrase = " ".join(f"action:{action}" for action in actions)
+
+    # adding lookback_period based filtering
+    date_filter = ""
+    if lookback_period:
+        lookback_period = validate_lookback_period(lookback_period)
+        date_filter = (
+            f"created:>={(datetime.now(tz=UTC) - relativedelta(**lookback_period))
+            .strftime('%Y-%m-%d')}"
+            if lookback_period
+            else ""
+        )
+
+    # adding actor-based filtering
+    actors_phrase = ""
+    if actors:
+        actors_phrase = " ".join(f"actor:{actor}" for actor in actors)
+
+    # adding exclude_actors based filtering
+    exclude_actors_phrase = ""
+    if exclude_actors:
+        exclude_actors_phrase = " ".join(f"-actor:{actor}" for actor in exclude_actors)
+    return " ".join(
+        section
+        for section in [
+            actions_phrase,
+            date_filter,
+            actors_phrase,
+            exclude_actors_phrase,
+        ]
+        if section
+    ).strip()
+
+
 class GithubRestApiClient:
     def __init__(
         self,
@@ -240,6 +302,13 @@ async def _get_paginated(
             query_params.update(params)
 
         while url is not None:
+            if "&page=100" in url:
+                logger.warning(
+                    "The GithubAPI has reached the maximum page size "
+                    "of 100. The returned data may be incomplete for request: %s",
+                    url,
+                )
+
             response = await self._get_retrying(
                 url, headers=headers, params=query_params
             )
@@ -331,23 +400,24 @@ async def fetch_all_organizations(self) -> AsyncGenerator[types.GithubOrg]:
             _fetch_problem("all organizations", e)
 
     async def fetch_enterprise_audit_log(
-        self, enterprise_name: str, actions: list[str], lookback_period: dict[str, int]
+        self,
+        enterprise_name: str,
+        actions: list[str],
+        actors: list[str],
+        exclude_actors: list[str],
+        lookback_period: dict[str, int],
     ) -> AsyncGenerator[types.GithubAuditLog]:
         """Fetches enterprise-wide audit log data
 
         https://docs.github.com/en/enterprise-cloud@latest/rest/enterprise-admin/audit-log?apiVersion=2022-11-28#get-the-audit-log-for-an-enterprise
         """
         try:
-            # adding action-based filtering
-            actions_phrase = " ".join(f"action:{action}" for action in actions)
-            # adding lookback_period based filtering
-            date_filter = (
-                f" created:>={(datetime.now(tz=UTC) - relativedelta(**lookback_period))
-                .strftime('%Y-%m-%d')}"
-                if lookback_period
-                else ""
+            search_phrase = build_search_phrase(
+                actions=actions,
+                actors=actors,
+                exclude_actors=exclude_actors,
+                lookback_period=lookback_period,
             )
-            search_phrase = f"{actions_phrase}{date_filter}"
 
             params = {"phrase": search_phrase} if search_phrase else {}
 

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nodestream-plugin-github"
-version = "0.14.1-beta.3"
+version = "0.14.1-beta.4"
 description = ""
 authors = [
     "Jon Bristow <[email protected]>",
@@ -17,6 +17,7 @@ nodestream = "^0.14"
 limits = "5.2.0"
 tenacity = "^9.0.0"
 httpx = ">=0.27,<0.28"
+freezegun = "^1.5.4"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.11.0"

pyproject.toml

@@ -85,6 +85,45 @@ async def test_pagination(httpx_mock: HTTPXMock):
     assert items == ["a", "b", "c", "d"]
 
 
+@pytest.mark.asyncio
+async def test_pagination_truncate_warning(
+    httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture
+):
+    client = GithubRestApiClient(
+        auth_token="test-auth-token",
+        github_hostname=DEFAULT_HOSTNAME,
+        user_agent="test-user-agent",
+        max_retries=0,
+        per_page=2,
+    )
+
+    next_page = f'<{DEFAULT_BASE_URL}/example?per_page=2&page=100>; rel="next"'
+    first_page = f'<${DEFAULT_BASE_URL}/example?per_page=2&page=99>; rel="first"'
+    httpx_mock.add_response(
+        url=f"{DEFAULT_BASE_URL}/example?per_page=2",
+        json=["a", "b"],
+        is_reusable=False,
+        headers={"link": f"{next_page}, {first_page}"},
+    )
+    httpx_mock.add_response(
+        url=f"{DEFAULT_BASE_URL}/example?per_page=2&page=100",
+        json=["c", "d"],
+        is_reusable=False,
+    )
+
+    with caplog.at_level("WARNING"):
+        items = [item async for item in client._get_paginated("example")]
+
+    assert items == ["a", "b", "c", "d"]
+
+    # Test that the warning message was logged
+    expected_warning = (
+        "The GithubAPI has reached the maximum page size of 100. "
+        "The returned data may be incomplete"
+    )
+    assert expected_warning in caplog.text
+
+
 def test_all_null_args():
     # noinspection PyTypeChecker
     assert GithubRestApiClient(auth_token=None, github_hostname=None)

tests/client/test_githubclient.py

@@ -1,5 +1,5 @@
 # noinspection PyProtectedMember
-from typing import Any
+from typing import Any, Optional
 
 from pytest_httpx import HTTPXMock
 
@@ -192,9 +192,9 @@ def get_repos_for_user(
             **kwargs,
         )
 
-    def get_enterprise_audit_logs(self, **kwargs: Any):
-        url = (
-            f"{self.base_url}/enterprises/test-enterprise"
-            f"/audit-log?per_page=100&phrase=action:protected_branch.create"
-        )
+    def get_enterprise_audit_logs(self, *, search_phrase: Optional[str], **kwargs: Any):
+        url = f"{self.base_url}/enterprises/test-enterprise/audit-log"
+        url += f"?per_page={self.per_page}"
+        if search_phrase:
+            url += f"&phrase={search_phrase}"
         self.add_response(url=url, **kwargs)