audit-labs/audit-tools

  1"""
  2Collect GitHub audit log events.
  3
  4Requires GitHub Enterprise Cloud. Skips gracefully with a warning if not
  5available.
  6"""
  7
  8import json
  9import sys
 10from datetime import date, datetime, timedelta, timezone
 11
 12import requests
 13
 14# GitHub audit logs retain non-Git events for 180 days. Supplying a created:
 15# qualifier is required to get events older than the default three-month window.
 16DEFAULT_LOOKBACK_DAYS = 180
 17DEFAULT_ACTION_FAMILIES = ["protected_branch", "repository_ruleset"]
 18DETAIL_FIELDS = [
 19    "name",
 20    "old_name",
 21    "branch",
 22    "repo",
 23    "repository",
 24    "operation_type",
 25    "ruleset_id",
 26    "ruleset_name",
 27    "ruleset_old_name",
 28    "ruleset_enforcement",
 29    "ruleset_old_enforcement",
 30    "ruleset_source_type",
 31    "ruleset_bypass_actors",
 32    "ruleset_bypass_actors_added",
 33    "ruleset_bypass_actors_deleted",
 34    "ruleset_bypass_actors_updated",
 35    "ruleset_conditions",
 36    "ruleset_conditions_added",
 37    "ruleset_conditions_deleted",
 38    "ruleset_conditions_updated",
 39    "ruleset_rules",
 40    "ruleset_rules_added",
 41    "ruleset_rules_deleted",
 42    "ruleset_rules_updated",
 43    "required_status_checks_enforcement_level",
 44    "strict_required_status_checks_policy",
 45    "pull_request_reviews_enforcement_level",
 46    "required_approving_review_count",
 47    "require_code_owner_review",
 48    "require_last_push_approval",
 49    "admin_enforced",
 50    "allow_force_pushes_enforcement_level",
 51    "allow_deletions_enforcement_level",
 52    "lock_branch_enforcement_level",
 53    "linear_history_requirement_enforcement_level",
 54    "signature_requirement_enforcement_level",
 55    "merge_queue_enforcement_level",
 56]
 57
 58
 59def audit_log(org, cfg, actions=None, lookback_days=DEFAULT_LOOKBACK_DAYS):
 60    """
 61    Return branch protection and repository ruleset audit events.
 62
 63    Returns an empty list with a warning if the org is not on GitHub Enterprise
 64    Cloud or the token does not have audit-log access.
 65    """
 66    action_families = actions or DEFAULT_ACTION_FAMILIES
 67    url = f"https://api.github.com/orgs/{org}/audit-log"
 68    since = date.today() - timedelta(days=lookback_days)
 69
 70    try:
 71        events = []
 72        for action_family in action_families:
 73            params = {
 74                "phrase": f"action:{action_family} created:>={since.isoformat()}",
 75                "include": "web",
 76                "order": "desc",
 77                "per_page": 100,
 78            }
 79            events.extend(_paginate_audit_log(url, cfg, params))
 80    except Exception as e:
 81        if "403" in str(e) or "404" in str(e):
 82            print(
 83                "Warning: audit log requires GitHub Enterprise Cloud, organization owner access, and audit-log token permissions -- skipping.",
 84                file=sys.stderr,
 85            )
 86            return []
 87        raise
 88
 89    rows = []
 90    for e in _dedupe_events(events):
 91        rows.append(
 92            {
 93                "action": e.get("action", ""),
 94                "actor": e.get("actor", ""),
 95                "repo": e.get("repo", ""),
 96                "branch_or_pattern": _branch_or_pattern(e),
 97                "operation_type": e.get("operation_type", ""),
 98                "summary": _event_summary(e),
 99                "details": _event_details(e),
100                "created_at": _format_created_at(e.get("created_at", "")),
101                "org": e.get("org", ""),
102            }
103        )
104    return rows
105
106
107def _paginate_audit_log(url, cfg, params):
108    """Fetch all audit-log cursor pages by following GitHub's Link header."""
109    results = []
110    next_url = url
111    next_params = params
112
113    while next_url:
114        resp = requests.get(
115            next_url,
116            headers=cfg["headers"],
117            params=next_params,
118            timeout=cfg["timeout"],
119        )
120        resp.raise_for_status()
121        data = resp.json()
122        if not data:
123            break
124
125        results.extend(data)
126        next_url = resp.links.get("next", {}).get("url")
127        next_params = None
128
129    return results
130
131
132def _event_details(event):
133    """Compact the change-specific audit-log fields into one CSV column."""
134    details = {
135        field: event[field]
136        for field in DETAIL_FIELDS
137        if field in event and event[field] not in (None, "")
138    }
139    return json.dumps(details, sort_keys=True)
140
141
142def _dedupe_events(events):
143    """Return events once, sorted newest first."""
144    seen = set()
145    unique = []
146    for event in events:
147        key = (
148            event.get("@timestamp") or event.get("created_at"),
149            event.get("action"),
150            event.get("actor"),
151            event.get("repo"),
152            event.get("operation_type"),
153            event.get("ruleset_id"),
154            event.get("name") or event.get("ruleset_name"),
155        )
156        if key in seen:
157            continue
158        seen.add(key)
159        unique.append(event)
160    return sorted(unique, key=_event_sort_value, reverse=True)
161
162
163def _branch_or_pattern(event):
164    """Find the most useful target label for branch and ruleset audit events."""
165    return (
166        event.get("name")
167        or event.get("branch")
168        or event.get("ruleset_name")
169        or event.get("ruleset_old_name")
170        or ""
171    )
172
173
174def _event_summary(event):
175    """Build a short human-readable evidence summary for the CSV."""
176    action = event.get("action", "")
177    operation = event.get("operation_type", "")
178    repo = event.get("repo", "")
179    target = _branch_or_pattern(event)
180
181    if action.startswith("repository_ruleset."):
182        source_type = event.get("ruleset_source_type", "")
183        scope = f"{source_type.lower()} " if source_type else ""
184        label = f" '{target}'" if target else ""
185        location = f" for {repo}" if repo else ""
186        return f"{operation or action} {scope}ruleset{label}{location}".strip()
187
188    if action.startswith("protected_branch."):
189        label = f" '{target}'" if target else ""
190        location = f" in {repo}" if repo else ""
191        return f"{operation or action} branch protection{label}{location}".strip()
192
193    return action
194
195
196def _format_created_at(value):
197    """Normalize GitHub audit-log timestamps to ISO-8601 UTC strings."""
198    if value in (None, ""):
199        return ""
200    if isinstance(value, (int, float)):
201        timestamp = value / 1000 if value > 9999999999 else value
202        return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()
203    return str(value)
204
205
206def _event_sort_value(event):
207    value = event.get("@timestamp") or event.get("created_at") or 0
208    if isinstance(value, (int, float)):
209        return value
210    try:
211        return datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp()
212    except ValueError:
213        return 0