Feat/audit logs v2

CHANGELOG.md

@@ -1,5 +1,31 @@
 # Change Log (v2.8.1+)
 
+## v4.7.0 [2026-07-31]
+
+__What's New:__
+
+* Migrated `audit_logs.logs` to the Audit Log API v2 (`/api/logs/v2`). Audit Log API v1 is deprecated and scheduled for retirement in mid-October 2026. No method signatures changed, so a simple SDK upgrade is all that is required.
+
+__Enhancements:__
+
+* `audit_logs.logs.query` now uses v2 token based pagination via the `next-page` response header, which resolves the prior issue where the final page of results could be omitted.
+* `audit_logs.logs.query` `from_time`/`to_time` now accept a string or int in addition to a `datetime`. Non-`datetime` values are passed through to the audit API, which supports ISO-8601 timestamps, epoch seconds/milliseconds, and relative expressions such as `now`, `yesterday`, or `1 day ago`. `datetime` inputs continue to be sent as UTC ISO-8601, so existing usage is unchanged.
+* `audit_logs.logs.[fields|operators]` now target the v2 endpoints.
+
+__Bug Fixes:__
+
+* None
+
+__Dependencies:__
+
+* None
+
+__Other:__
+
+* New errors that can be raised by `audit_logs.logs.query`:
+  * `exceptions.AuditLogCsvDownloadError` — when `csv=True` and the presigned S3 URL for the CSV export cannot be downloaded (e.g. the execution environment lacks outbound HTTPS access to AWS S3).
+  * `exceptions.InvalidRequest` — when the requested time frame exceeds the v2 maximum of 7 days (the human-readable API validation message is surfaced).
+
 ## v4.6.1 [2026-07-07]
 
 __Enhancements:__

src/britive/__init__.py

@@ -1 +1 @@
-__version__ = '4.6.1'
+__version__ = '4.7.0'

src/britive/audit_logs/logs.py

@@ -1,11 +1,15 @@
 from datetime import datetime, timedelta, timezone
-from typing import Any
+from typing import Any, Union
+
+import requests
+
+from ..exceptions import AuditLogCsvDownloadError
 
 
 class Logs:
     def __init__(self, britive) -> None:
         self.britive = britive
-        self.base_url = f'{self.britive.base_url}/logs'
+        self.base_url = f'{self.britive.base_url}/logs/v2'
 
     def fields(self) -> dict:
         """
@@ -26,7 +30,11 @@ def operators(self) -> dict:
         return self.britive.get(f'{self.base_url}/operators')
 
     def query(
-        self, from_time: datetime = None, to_time: datetime = None, filter_expression: str = None, csv: bool = False
+        self,
+        from_time: Union[datetime, str, int] = None,
+        to_time: Union[datetime, str, int] = None,
+        filter_expression: str = None,
+        csv: bool = False,
     ) -> Any:
         """
         Retrieve audit log events.
@@ -36,34 +44,72 @@ def query(
             - True: A CSV string is returned. The caller must persist the CSV string to disk.
             - False: A python list of audit events is returned.
 
+        Both `from_time` and `to_time` accept either a `datetime.datetime` object or a value understood directly
+        by the audit API. Accepted API values include ISO-8601 timestamps, epoch seconds, epoch milliseconds, and
+        relative expressions such as `now`, `yesterday`, or `1 day ago`. When a value without a timezone is provided
+        the API treats it as UTC; when a timezone offset is provided the API honors it.
+
         :param from_time: Lower end of the time frame to search. If not provided will default to
-            7 days before `to_time`. `from_time` will be interpreted as if in UTC timezone so it is up to the caller to
-            ensure that the datetime object represents UTC. No timezone manipulation will occur.
+            7 days before `to_time`. A `datetime` object will be interpreted as if in UTC timezone so it is up to the
+            caller to ensure that the datetime object represents UTC (no timezone manipulation is performed). A string
+            or int is passed through to the audit API as-is (e.g. `'1 day ago'`, `'2026-06-01T00:00:00Z'`, epoch).
         :param to_time: Upper end of the time frame to search. If not provided will default to
-            `datetime.datetime.utcnow()`. `to_time` will be interpreted as if in UTC timezone so it is up to the caller
-            to ensure that the datetime object represents UTC. No timezone manipulation will occur.
+            `datetime.datetime.now(timezone.utc)`. A `datetime` object will be interpreted as if in UTC timezone so it
+            is up to the caller to ensure that the datetime object represents UTC (no timezone manipulation is
+            performed). A string or int is passed through to the audit API as-is.
+            Note: the audit API allows a maximum time frame of 7 days between `from_time` and `to_time`.
         :param filter_expression: The expression used to filter the results. A list of available fields and operators
             can be found using `britive.audit_logs.logs.fields` and `britive.audit_logs.logs.operators`, respectively.
             Multiple filter expressions must be joined together by `and`. No other join operator is support.
             Example: actor.displayName co "bob" and event.displayName eq "application"
         :param csv: Will result in a CSV string of the audit events being returned instead of a python list of events.
         :return: Either python list of events (dicts) or CSV string.
-        :raises: ValueError - If from_time is greater than to_time.
+        :raises: ValueError - If both `from_time` and `to_time` are `datetime` objects and `from_time` is greater
+            than `to_time`. When either bound is a string/int it is left to the audit API to validate.
         """
 
-        to_time = to_time or datetime.now(timezone.utc)
-        from_time = from_time or to_time - timedelta(days=7)
+        if to_time is None:
+            to_time = datetime.now(timezone.utc)
+        if from_time is None:
+            # default the lower bound to 7 days before the upper bound when the upper bound is a datetime;
+            # otherwise (a passed-through string/int) fall back to 7 days before now.
+            base = to_time if isinstance(to_time, datetime) else datetime.now(timezone.utc)
+            from_time = base - timedelta(days=7)
 
-        if from_time > to_time:
+        # only enforce ordering when both bounds are datetimes; string/int values are validated by the audit API
+        if isinstance(from_time, datetime) and isinstance(to_time, datetime) and from_time > to_time:
             raise ValueError('from_time must occur before to_time.')
 
-        params = {
-            'from': from_time.isoformat(sep='T', timespec='seconds').split('+')[0] + 'Z',
-            'to': to_time.isoformat(sep='T', timespec='seconds').split('+')[0] + 'Z',
-        }
+        # a datetime is sent as a UTC ISO-8601 timestamp with a trailing `Z`; a string/int (ISO, epoch, or a
+        # relative expression like '1 day ago') is passed through untouched for the audit API to interpret.
+        def fmt(value):
+            if isinstance(value, datetime):
+                return value.isoformat(sep='T', timespec='seconds').split('+')[0] + 'Z'
+            return value
+
+        params = {'from': fmt(from_time), 'to': fmt(to_time)}
         if filter_expression:
             params['filter'] = filter_expression
-        if not csv:
-            params['size'] = 200
 
-        return self.britive.get(f'{self.base_url}{"/csv" if csv else ""}', params=params)
+        if csv:
+            return self._download_csv(params)
+
+        params['size'] = 200
+        return self.britive.get(self.base_url, params=params)
+
+    def _download_csv(self, params: dict) -> str:
+        # The v2 CSV export endpoint does not return the CSV content directly. Instead it returns a presigned
+        # S3 download URL. To preserve the v1 behavior of returning the CSV as a string, download the file from
+        # the presigned URL and return its content. The presigned URL is self-contained and must be requested
+        # without the Britive authorization header (S3 rejects unexpected headers), so a bare request is used.
+        details = self.britive.get(f'{self.base_url}/csv', params=params)
+        try:
+            response = requests.get(details['downloadUrl'], timeout=60)
+            response.raise_for_status()
+        except requests.exceptions.RequestException as e:
+            raise AuditLogCsvDownloadError(
+                'Retrieved the audit log CSV download URL from the Britive tenant but failed to download the file '
+                'from the presigned S3 URL. Ensure the execution environment has outbound HTTPS access to AWS S3 '
+                f'(*.s3.amazonaws.com). Underlying error: {e}'
+            ) from e
+        return response.content.decode('utf-8')

src/britive/britive.py

@@ -290,6 +290,16 @@ def __request(self, method, url, params=None, data=None, json=None, headers=None
                     break
                 url = response.headers['next-page']
                 params = {}
+            elif _pagination_type == 'audit_v2':
+                # v2 audit logs return a `records` list and, when more data exists, a page token in the `next-page`
+                # response header. Unlike the `audit` type above - which treats the `next-page` header as a full URL
+                # to follow - this token is an opaque cursor passed back as the `pageToken` query parameter against
+                # the same URL.
+                return_data += result.get('records', [])
+                next_page_token = response.headers.get('next-page')
+                if not next_page_token:
+                    break
+                params['pageToken'] = next_page_token
             elif _pagination_type == 'secmgr':
                 return_data += result['result']
                 url = result['pagination'].get('next', '')

src/britive/exceptions/__init__.py

@@ -22,6 +22,10 @@ class ApprovalWorkflowTimedOut(BritiveException):
     pass
 
 
+class AuditLogCsvDownloadError(BritiveException):
+    pass
+
+
 class Conflict(BritiveException):
     pass
 

src/britive/helpers/utils.py

@@ -53,6 +53,8 @@ def pagination_type(headers, result) -> str:
 
     if is_dict and all(x in result for x in ('count', 'page', 'size', 'data')):
         return 'inline'
+    if is_dict and 'records' in result:  # this is how audit_logs.query() (v2) paginates - via a page token
+        return 'audit_v2'
     if is_dict and has_next_page_header and all(x in result for x in ('data', 'reportId')):  # reports
         return 'report'
     if has_next_page_header:  # this interesting way of paginating is how audit_logs.query() does it