audit-labs/audit-tools
A collection of scripts, queries, and other goodies you can use in an audit.
clone: git clone https://gitbay.org/audit-labs/audit-tools.git
a877dc7b44b76e85fce1338ab1d2a11c7769018e
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-07-29T04:13:39Z
README.org | 13 +- applications/__init__.py | 0 applications/github/__init__.py | 0 applications/gitlab/README.md | 186 +++--------- applications/gitlab/__init__.py | 0 applications/gitlab/approvals.py | 38 --- applications/gitlab/audit.py | 149 +++++++++ applications/gitlab/branch_protections.py | 25 -- applications/gitlab/collectors/__init__.py | 0 applications/gitlab/collectors/api.py | 41 +++ applications/gitlab/collectors/approvals.py | 48 +++ applications/gitlab/collectors/audit_events.py | 46 +++ .../gitlab/collectors/branch_protections.py | 44 +++ applications/gitlab/collectors/members.py | 72 +++++ applications/gitlab/collectors/pipelines.py | 38 +++ applications/gitlab/collectors/projects.py | 33 ++ applications/gitlab/collectors/settings.py | 38 +++ applications/gitlab/config.py | 45 +++ applications/gitlab/passwords.py | 39 --- applications/gitlab/pipelines.py | 59 ---- applications/gitlab/provisioning.py | 32 -- applications/gitlab/reporters/__init__.py | 0 applications/gitlab/reporters/csv_reporter.py | 46 +++ applications/gitlab/repositories.py | 51 ---- applications/gitlab/users.py | 53 ---- audit_tui.py | 7 + conftest.py | 10 + requirements.txt | 1 + tui/README.md | 63 ++++ tui/__init__.py | 1 + tui/app.py | 337 +++++++++++++++++++++ tui/common.py | 30 ++ tui/github_runner.py | 180 +++++++++++ tui/gitlab_runner.py | 163 ++++++++++ tui/platforms.py | 144 +++++++++ tui/tests/test_app.py | 127 ++++++++ tui/tests/test_github_runner.py | 114 +++++++ tui/tests/test_gitlab_runner.py | 113 +++++++ 38 files changed, 1944 insertions(+), 442 deletions(-) @@ -13,7 +13,7 @@ specific audit environments. |----------------------+------------------------------------------------------------------------------| | =applications/aws/= | AWS IAM users, password policy, and S3 bucket analysis | | =applications/github/= | GitHub admin enumeration, audit log, branch protections, and commit analysis | -| =applications/gitlab/= | GitLab user provisioning, branch protections, approvals, pipelines, and more | +| =applications/gitlab/= | GitLab group/project members, branch protections, approvals, pipelines, audit events | | =databases/mongo/= | MongoDB admin enumeration | | =databases/mysql/= | MySQL admin and password queries | | =databases/oracle/= | Oracle admin queries | @@ -22,6 +22,7 @@ specific audit environments. | =os/linux/= | Linux OS reporting, password file analysis, and SSH root login checks | | =project_management/= | Audit project tracking dashboards (Alteryx, Dash, Power BI) | | =sampling/= | Random and stratified sampling tools | +| =tui/= | Interactive terminal UI that walks you through running an audit | ** Getting Started @@ -57,6 +58,16 @@ python sampling/sample.py Output will be shown in the terminal or saved to a file, depending on the script. +*Interactive TUI* + +To pick a platform and be walked through an audit interactively: + +#+begin_src bash +python audit_tui.py +#+end_src + +See =tui/README.md= for details. GitHub and GitLab are supported. + ** Contributing Contributions are welcome. You can contribute by: new file mode 100644 new file mode 100644 @@ -1,160 +1,58 @@ -# `approvals.py` +> **NOTE**: The token used across all collectors needs at least the `read_api` +> scope. Some checks need more: +> - **Approval rules** and **audit events** require a GitLab Premium or Ultimate +> subscription. +> - **Password policy** reads instance application settings, which require an +> admin token on a self-hosted instance (not available on GitLab.com). +> +> Checks that are unavailable are skipped with a warning; the rest still run. -\\This script requires an active Premium or Ultimate subscription.\*\\ +--- -``` bash -python ./approvals.py -``` +# `audit.py` — Unified GitLab Audit Tool -``` text -Rule: All Members - Approvals Required: 1 - Rule type: any_approver -Rule: Default - Approvals Required: 1 - Rule type: regular - Protected Branch: master - Eligible Approver: Christian Cleberg -``` +Runs all collectors against a GitLab group (including its subgroups) and writes +a timestamped audit package to disk. -# `branch_protections.py` +## Setup -``` bash -python ./branch_protections.py +```bash +export GITLAB_TOKEN=your_token +export GITLAB_GROUP=your_group_id_or_path +# Self-hosted only: +export GITLAB_URL=https://gitlab.example.com/api/v4 ``` -``` json -[ - { - "id": 148448212, - "name": "main", - "push_access_levels": [ - { - "id": 185900194, - "access_level": 40, - "access_level_description": "Maintainers", - "deploy_key_id": null, - "user_id": null, - "group_id": null - } - ], - "merge_access_levels": [ - { - "id": 156461000, - "access_level": 40, - "access_level_description": "Maintainers", - "user_id": null, - "group_id": null - } - ], - "allow_force_push": false, - "unprotect_access_levels": [], - "code_owner_approval_required": false, - "inherited": false - } -] -``` +## Usage -# `passwords.py` +```bash +# Basic run — uses GITLAB_TOKEN and GITLAB_GROUP from environment +python audit.py -**This script does not apply to GitLab.com. This is for self-hosted -instances only.** +# Override group, set output directory +python audit.py --group my-group --out ./output -``` bash -python ./passwords.py +# Point at a self-hosted instance +python audit.py --url https://gitlab.example.com/api/v4 ``` -``` text -# TODO: Need access to a self-hosted version of GitLab to test this out. -``` +The group may be a numeric ID (`1234567`) or a URL path (`my-group/sub-group`). -# `pipelines.py` +## Output -``` bash -python ./pipelines.py -``` - -``` text -Pipeline ID: 1754222228 - Status: failed - Ref: master - Created At: 2025-04-06T03:39:15.065Z - Duration: N/A seconds - Configuration: N/A -Pipeline ID: 1754221831 - Status: failed - Ref: pr-1 - Created At: 2025-04-06T03:37:42.333Z - Duration: N/A seconds - Configuration: N/A -Pipeline ID: 1754220271 - Status: failed - Ref: pr-1 - Created At: 2025-04-06T03:33:38.606Z - Duration: N/A seconds - Configuration: N/A -Pipeline ID: 1754214637 - Status: failed - Ref: master - Created At: 2025-04-06T03:21:39.902Z - Duration: N/A seconds - Configuration: N/A -``` - -# `provisioning.py` - -\\This script requires an active Premium or Ultimate subscription.\*\\ - -``` bash -python ./provisioning.py -``` - -``` text -Group: 105300140 - 2025-04-08T03:33:17.055Z : Action: member_created, Member: 128029250, Author: 24608590 -``` +Creates a directory: `<out>/gitlab_audit_<group>_<YYYY-MM-DD>/` -# `repositories.py` +| File | Contents | +|---|---| +| `group_members.csv` | Group members with access level and role | +| `projects.csv` | All projects in the group and subgroups | +| `project_members.csv` | Members and access levels for every project | +| `branch_protections.csv` | Protected-branch settings across all projects | +| `pipelines.csv` | CI/CD pipeline history across all projects | +| `approval_rules.csv` | Merge-request approval rules (Premium/Ultimate) | +| `audit_events.csv` | Group membership audit events (Premium/Ultimate) | +| `password_policy.csv` | Instance password policy (self-hosted, admin token) | +| `summary.txt` | Row counts per section | -``` shell -python ./repositories.py -``` - -``` text -# User ID Example -Projects under ID: ccleberg: -- audit-tools (ID: 68757698) -- cleberg.net (ID: 68701468) - -# Group ID Example -Projects under ID: phryq: -- Yoshi Cli (ID: 68757750) -- pages-demo (ID: 68757186) -``` - -# `users.py` - -``` bash -python ./users.py -``` - -``` text -Access Level Roles: - 0 : No access - 5 : Minimal access - 10 : Guest - 15 : Planner - 20 : Reporter - 30 : Developer - 40 : Maintainer - 50 : Owner - 60 : Admin - - -Group 97083755 Members: -Username: ccleberg, Access Level: 50 - -Project 68701468 Members: -Username: ccleberg, Access Level: 50 -Username: project_68701468_bot_2c7ee010a479c0e48cdb4c7c5cfae886, Access Level: 40 -``` +The per-project checks reuse a single enumeration of the group's projects, so +the group is listed only once per run. new file mode 100644 deleted file mode 100644 @@ -1,38 +0,0 @@ -""" -Extract merge request approval rules and their statuses in GitLab. -""" - -import requests - -BASE_URL = "https://gitlab.com/api/v4" -PRIVATE_TOKEN = "your_access_token" -PROJECT_ID = "your_project_id" -TIMEOUT = 30 - -URL = f"{BASE_URL}/projects/{PROJECT_ID}/approval_rules" -HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN} - -if __name__ == "__main__": - # Get approval rules - response = requests.get(URL, headers=HEADERS, timeout=TIMEOUT) - if response.status_code == 200: - approval_rules = response.json() - for rule in approval_rules: - name = rule["name"] - approvals_required = rule["approvals_required"] - rule_type = rule["rule_type"] - protected_branches = rule["protected_branches"] - eligible_approvers = rule["eligible_approvers"] - print(f"Rule: {name}") - print(f" Approvals Required: {approvals_required}") - print(f" Rule type: {rule_type}") - for branch in protected_branches: - branch_name = branch["name"] - print(f" Protected Branch: {branch_name}") - for approver in eligible_approvers: - approver_username = approver["name"] - print(f" Eligible Approver: {approver_username}") - else: - print( - f"Failed to fetch approval rules: {response.status_code}, {response.text}" - ) new file mode 100644 @@ -0,0 +1,149 @@ +""" +GitLab audit CLI. + +Runs all collectors against a GitLab group and writes a timestamped audit +package to an output directory. + +Usage: + export GITLAB_TOKEN=your_token + export GITLAB_GROUP=your_group_id_or_path + + python audit.py + python audit.py --group my-group + python audit.py --group my-group --out ./output + python audit.py --group my-group --url https://gitlab.example.com/api/v4 + +Output: + <out>/gitlab_audit_<group>_<date>/ + group_members.csv + projects.csv + project_members.csv + branch_protections.csv + pipelines.csv + approval_rules.csv + audit_events.csv + password_policy.csv + summary.txt +""" + +import argparse +import os +import sys +from datetime import date + +import config +from collectors import ( + approvals, + audit_events, + branch_protections, + members, + pipelines, + projects, + settings, +) +from reporters import csv_reporter + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Generate a GitLab audit package for a group." + ) + parser.add_argument( + "--group", + help="GitLab group ID or path. Overrides GITLAB_GROUP env var.", + ) + parser.add_argument( + "--url", + help="GitLab API base URL. Overrides GITLAB_URL env var. " + "Default: https://gitlab.com/api/v4", + ) + parser.add_argument( + "--out", + default="./output", + help="Directory to write the audit package into. Default: ./output", + ) + return parser.parse_args() + + +def run(): + args = parse_args() + cfg = config.load(group_override=args.group, base_url_override=args.url) + group = cfg["group"] + + safe_group = group.replace("/", "-") + output_dir = os.path.join( + args.out, f"gitlab_audit_{safe_group}_{date.today().isoformat()}" + ) + + print(f"GitLab Audit — {group}") + print(f"Output directory: {output_dir}") + print() + + sections = [] + + def collect(label, fn, filename, *fn_args): + print(f"Collecting: {label}...") + try: + rows = fn(*fn_args) + except Exception as e: + print(f" Error: {e}", file=sys.stderr) + rows = [] + csv_reporter.write(output_dir, filename, rows) + sections.append((label, len(rows))) + return rows + + print("Enumerating projects (shared cache)...") + try: + project_cache = projects.fetch_projects(group, cfg) + except Exception as e: + print(f" Error enumerating projects: {e}", file=sys.stderr) + project_cache = [] + + collect("Group members", members.group_members, "group_members.csv", group, cfg) + collect( + "Projects", projects.project_list, "projects.csv", group, cfg, project_cache + ) + collect( + "Project members", + members.project_members, + "project_members.csv", + group, + cfg, + project_cache, + ) + collect( + "Branch protections", + branch_protections.branch_protections, + "branch_protections.csv", + group, + cfg, + project_cache, + ) + collect( + "Pipelines", pipelines.pipelines, "pipelines.csv", group, cfg, project_cache + ) + collect( + "Approval rules", + approvals.approval_rules, + "approval_rules.csv", + group, + cfg, + project_cache, + ) + collect("Audit events", audit_events.audit_events, "audit_events.csv", group, cfg) + collect( + "Password policy", + settings.password_policy, + "password_policy.csv", + group, + cfg, + ) + + print() + csv_reporter.write_summary(output_dir, group, sections) + print() + print("Done.") + + +if __name__ == "__main__": + run() deleted file mode 100644 @@ -1,25 +0,0 @@ -""" -List all branch protection rules and their configurations in GitLab. -""" - -import requests -import json - -BASE_URL = "https://gitlab.com/api/v4" -PRIVATE_TOKEN = "your_access_token" -PROJECT_ID = "your_project_id" -TIMEOUT = 30 - -URL = f"{BASE_URL}/projects/{PROJECT_ID}/protected_branches" -HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN} - -if __name__ == "__main__": - # Get protected branches - response = requests.get(URL, headers=HEADERS, timeout=TIMEOUT) - if response.status_code == 200: - protected_branches = response.json() - print(json.dumps(protected_branches, indent=4)) - else: - print( - f"Failed to fetch protected branches: {response.status_code}, {response.text}" - ) new file mode 100644 new file mode 100644 @@ -0,0 +1,41 @@ +"""Shared GitLab API helpers.""" + +from urllib.parse import quote + +import requests + +DEFAULT_BASE_URL = "https://gitlab.com/api/v4" + + +def enc(value): + """URL-encode a group or project identifier. + + GitLab accepts either a numeric ID or a URL-encoded path (e.g. + ``my-group/sub-group``). Numeric IDs pass through unchanged. + """ + return quote(str(value), safe="") + + +def paginate(url, cfg, params=None): + """Fetch all pages from a GitLab endpoint using the X-Next-Page header.""" + results = [] + p = dict(params or {}) + p["per_page"] = 100 + page = 1 + + while True: + p["page"] = page + resp = requests.get( + url, headers=cfg["headers"], params=p, timeout=cfg["timeout"] + ) + resp.raise_for_status() + data = resp.json() + if not data: + break + results.extend(data) + next_page = resp.headers.get("X-Next-Page") + if not next_page: + break + page = int(next_page) + + return results new file mode 100644 @@ -0,0 +1,48 @@ +""" +Collect merge-request approval rules for every project in the group. + +Approval rules require a GitLab Premium or Ultimate subscription. Projects that +return 403/404 (feature unavailable) are skipped with a warning. +""" + +import sys + +import requests + +from .api import paginate + + +def approval_rules(group, cfg, projects): + rows = [] + for p in projects: + try: + rules = paginate( + f"{cfg['base_url']}/projects/{p['id']}/approval_rules", cfg + ) + except requests.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + print( + f" Skipping {p.get('path_with_namespace', p['id'])}: " + f"approval_rules returned {e.response.status_code}", + file=sys.stderr, + ) + continue + raise + for rule in rules: + approvers = ", ".join( + a.get("name", "") for a in rule.get("eligible_approvers", []) + ) + branches = ", ".join( + b.get("name", "") for b in rule.get("protected_branches", []) + ) + rows.append( + { + "project": p.get("path_with_namespace", ""), + "rule": rule.get("name", ""), + "rule_type": rule.get("rule_type", ""), + "approvals_required": rule.get("approvals_required", 0), + "protected_branches": branches or "(all)", + "eligible_approvers": approvers or "(none)", + } + ) + return rows new file mode 100644 @@ -0,0 +1,46 @@ +""" +Collect group membership audit events (created / updated / destroyed). + +Group audit events require a GitLab Premium or Ultimate subscription. Returns an +empty list with a warning if the endpoint is unavailable (403/404). +""" + +import sys + +import requests + +from .api import enc, paginate + +MEMBER_ACTIONS = {"member_created", "member_updated", "member_destroyed"} + + +def audit_events(group, cfg): + try: + events = paginate(f"{cfg['base_url']}/groups/{enc(group)}/audit_events", cfg) + except requests.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + print( + "Warning: group audit events require GitLab Premium/Ultimate and " + "owner access -- skipping.", + file=sys.stderr, + ) + return [] + raise + + rows = [] + for event in events: + action = event.get("event_name", "") + if action not in MEMBER_ACTIONS: + continue + details = event.get("details", {}) + rows.append( + { + "created_at": event.get("created_at", ""), + "action": action, + "member_id": details.get("member_id", ""), + "target": details.get("target_details", ""), + "author_id": event.get("author_id", ""), + "entity_type": event.get("entity_type", ""), + } + ) + return rows new file mode 100644 @@ -0,0 +1,44 @@ +"""Collect protected-branch settings for every project in the group.""" + +import sys + +import requests + +from .api import paginate + + +def _levels(entries): + """Summarize an access-level list (push/merge/unprotect) into one string.""" + return ", ".join(e.get("access_level_description", "") for e in entries) or "(none)" + + +def branch_protections(group, cfg, projects): + rows = [] + for p in projects: + try: + protected = paginate( + f"{cfg['base_url']}/projects/{p['id']}/protected_branches", cfg + ) + except requests.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + print( + f" Skipping {p.get('path_with_namespace', p['id'])}: " + f"protected_branches returned {e.response.status_code}", + file=sys.stderr, + ) + continue + raise + for b in protected: + rows.append( + { + "project": p.get("path_with_namespace", ""), + "branch": b.get("name", ""), + "push_access": _levels(b.get("push_access_levels", [])), + "merge_access": _levels(b.get("merge_access_levels", [])), + "allow_force_push": b.get("allow_force_push"), + "code_owner_approval_required": b.get( + "code_owner_approval_required" + ), + } + ) + return rows new file mode 100644 @@ -0,0 +1,72 @@ +""" +Collect group and project membership with access levels. + +GitLab access levels: + 0 No access 5 Minimal 10 Guest 15 Planner + 20 Reporter 30 Developer 40 Maintainer 50 Owner 60 Admin +""" + +import sys + +import requests + +from .api import enc, paginate + +ACCESS_LEVELS = { + 0: "No access", + 5: "Minimal", + 10: "Guest", + 15: "Planner", + 20: "Reporter", + 30: "Developer", + 40: "Maintainer", + 50: "Owner", + 60: "Admin", +} + + +def _role(level): + return ACCESS_LEVELS.get(level, str(level)) + + +def group_members(group, cfg): + """Group members, including those inherited from parent groups.""" + members = paginate(f"{cfg['base_url']}/groups/{enc(group)}/members/all", cfg) + return [ + { + "username": m["username"], + "name": m.get("name", ""), + "access_level": m["access_level"], + "role": _role(m["access_level"]), + "state": m.get("state", ""), + } + for m in members + ] + + +def project_members(group, cfg, projects): + """Direct and inherited members of every project in the group.""" + rows = [] + for p in projects: + try: + members = paginate(f"{cfg['base_url']}/projects/{p['id']}/members/all", cfg) + except requests.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + print( + f" Skipping {p.get('path_with_namespace', p['id'])}: " + f"members returned {e.response.status_code}", + file=sys.stderr, + ) + continue + raise + for m in members: + rows.append( + { + "project": p.get("path_with_namespace", ""), + "username": m["username"], + "name": m.get("name", ""), + "access_level": m["access_level"], + "role": _role(m["access_level"]), + } + ) + return rows new file mode 100644 @@ -0,0 +1,38 @@ +"""Collect CI/CD pipeline history for every project in the group.""" + +import sys + +import requests + +from .api import paginate + + +def pipelines(group, cfg, projects): + rows = [] + for p in projects: + try: + project_pipelines = paginate( + f"{cfg['base_url']}/projects/{p['id']}/pipelines", cfg + ) + except requests.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + print( + f" Skipping {p.get('path_with_namespace', p['id'])}: " + f"pipelines returned {e.response.status_code}", + file=sys.stderr, + ) + continue + raise + for pipe in project_pipelines: + rows.append( + { + "project": p.get("path_with_namespace", ""), + "pipeline_id": pipe.get("id"), + "status": pipe.get("status", ""), + "ref": pipe.get("ref", ""), + "source": pipe.get("source", ""), + "created_at": pipe.get("created_at", ""), + "web_url": pipe.get("web_url", ""), + } + ) + return rows new file mode 100644 @@ -0,0 +1,33 @@ +""" +Enumerate the projects in a GitLab group. + +fetch_projects() returns the raw project objects once; the per-project +collectors reuse that cache to avoid re-listing the group. +""" + +from .api import enc, paginate + + +def fetch_projects(group, cfg): + """List all projects in the group, including subgroups.""" + return paginate( + f"{cfg['base_url']}/groups/{enc(group)}/projects", + cfg, + {"include_subgroups": "true", "archived": "false"}, + ) + + +def project_list(group, cfg, projects): + """Format the project cache into audit rows.""" + return [ + { + "id": p["id"], + "name": p["name"], + "path": p.get("path_with_namespace", ""), + "visibility": p.get("visibility", ""), + "default_branch": p.get("default_branch", ""), + "archived": p.get("archived", False), + "web_url": p.get("web_url", ""), + } + for p in projects + ] new file mode 100644 @@ -0,0 +1,38 @@ +""" +Collect the instance password policy from application settings. + +Requires an admin token on a self-hosted instance; not available on +GitLab.com. Returns an empty list with a warning on 403/404. +""" + +import sys + +import requests + +PASSWORD_FIELDS = [ + "minimum_password_length", + "password_number_required", + "password_symbol_required", + "password_uppercase_required", + "password_lowercase_required", +] + + +def password_policy(group, cfg): + """group is unused; application settings are instance-wide.""" + url = f"{cfg['base_url']}/application/settings" + try: + resp = requests.get(url, headers=cfg["headers"], timeout=cfg["timeout"]) + resp.raise_for_status() + except requests.HTTPError as e: + if e.response is not None and e.response.status_code in (403, 404): + print( + "Warning: application settings require an admin token on a " + "self-hosted instance -- skipping.", + file=sys.stderr, + ) + return [] + raise + + settings = resp.json() + return [{field: settings.get(field, "Not set") for field in PASSWORD_FIELDS}] new file mode 100644 @@ -0,0 +1,45 @@ +""" +Configuration loader for the GitLab audit tool. + +Reads GITLAB_TOKEN, GITLAB_GROUP, and (optionally) GITLAB_URL from the +environment. + +Usage: + export GITLAB_TOKEN=your_token + export GITLAB_GROUP=your_group_id_or_path + export GITLAB_URL=https://gitlab.example.com/api/v4 # self-hosted only +""" + +import os +import sys + +from collectors.api import DEFAULT_BASE_URL + + +def load(group_override=None, base_url_override=None): + """Return a config dict. Exits with an error if required values are missing.""" + token = os.environ.get("GITLAB_TOKEN", "").strip() + group = group_override or os.environ.get("GITLAB_GROUP", "").strip() + base_url = ( + base_url_override + or os.environ.get("GITLAB_URL", "").strip() + or DEFAULT_BASE_URL + ) + + missing = [] + if not token: + missing.append("GITLAB_TOKEN") + if not group: + missing.append("GITLAB_GROUP (or pass --group)") + + if missing: + print(f"Error: missing required values: {', '.join(missing)}", file=sys.stderr) + sys.exit(1) + + return { + "token": token, + "group": group, + "base_url": base_url.rstrip("/"), + "headers": {"PRIVATE-TOKEN": token}, + "timeout": 30, + } deleted file mode 100644 @@ -1,39 +0,0 @@ -""" -Verify if password policies are enforced in a self-hosted GitLab instance. - - Ref: https://docs.gitlab.com/api/settings/ -""" - -import requests - -BASE_URL = "https://gitlab.com/api/v4" -PRIVATE_TOKEN = "your_access_token" -TIMEOUT = 30 - -URL = f"{BASE_URL}/application/settings" -HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN} - -if __name__ == "__main__": - # Get application settings - response = requests.get(URL, headers=HEADERS, timeout=TIMEOUT) - if response.status_code == 200: - settings = response.json() - minimum_password_length = settings.get("minimum_password_length", "Not set") - password_number_required = settings.get("password_number_required", "Not set") - password_symbol_required = settings.get("password_symbol_required", "Not set") - password_uppercase_required = settings.get( - "password_uppercase_required", "Not set" - ) - password_lowercase_required = settings.get( - "password_lowercase_required", "Not set" - ) - - print(f"Password Length: {minimum_password_length}") - print(f"Password Number Required: {password_number_required}") - print(f"Password Symbol Required: {password_symbol_required}") - print(f"Password Uppercase Required: {password_uppercase_required}") - print(f"Password Lowercase Required: {password_lowercase_required}") - else: - print( - f"Failed to fetch application settings: {response.status_code}, {response.text}" - ) deleted file mode 100644 @@ -1,59 +0,0 @@ -""" -Review CI/CD pipelines and their configurations for a specific GitLab project. -""" - -import requests - -BASE_URL = "https://gitlab.com/api/v4" -PRIVATE_TOKEN = "your_access_token" -PROJECT_ID = "project_id" -TIMEOUT = 30 - -HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN} - -if __name__ == "__main__": - page = 1 - per_page = 100 - - while True: - response = requests.get( - f"{BASE_URL}/projects/{PROJECT_ID}/pipelines", - headers=HEADERS, - params={"page": page, "per_page": per_page}, - timeout=TIMEOUT, - ) - if response.status_code == 200: - pipelines = response.json() - if not pipelines: - break - - for pipeline in pipelines: - pipeline_id = pipeline["id"] - status = pipeline["status"] - ref = pipeline["ref"] - created_at = pipeline["created_at"] - duration = pipeline.get("duration", "N/A") - - print(f"Pipeline ID: {pipeline_id}") - print(f" Status: {status}") - print(f" Ref: {ref}") - print(f" Created At: {created_at}") - print(f" Duration: {duration} seconds") - - detail_response = requests.get( - f"{BASE_URL}/projects/{PROJECT_ID}/pipelines/{pipeline_id}", - headers=HEADERS, - timeout=TIMEOUT, - ) - if detail_response.status_code == 200: - pipeline_details = detail_response.json() - print(f" Configuration: {pipeline_details.get('config', 'N/A')}") - else: - print( - f" Failed to fetch pipeline details: {detail_response.status_code}, {detail_response.text}" - ) - - page += 1 - else: - print(f"Failed to fetch pipelines: {response.status_code}, {response.text}") - break deleted file mode 100644 @@ -1,32 +0,0 @@ -""" -Track user creation and deletion events in GitLab with timestamps. -""" - -import requests - -BASE_URL = "https://gitlab.com/api/v4" -PRIVATE_TOKEN = "your_access_token" -GROUP_ID = "your_group_id" -TIMEOUT = 30 - -URL = f"{BASE_URL}/groups/{GROUP_ID}/audit_events" -HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN} - -if __name__ == "__main__": - # Get audit events - response = requests.get(URL, headers=HEADERS, timeout=TIMEOUT) - if response.status_code == 200: - audit_events = response.json() - for event in audit_events: - if event["entity_type"] == "User" or event["entity_type"] == "Group": - action = event["event_name"] - member_id = event["details"].get("member_id") - created_at = event["created_at"] - author = event["author_id"] - if action in ["member_created", "member_destroyed", "member_updated"]: - print( - f"Group: {GROUP_ID}\n", - f" {created_at} : Action: {action}, Member: {member_id}, Author: {author}", - ) - else: - print(f"Failed to fetch audit events: {response.status_code}, {response.text}") new file mode 100644 new file mode 100644 @@ -0,0 +1,46 @@ +"""CSV reporter: writes one CSV file per data section into an output directory.""" + +import csv +import os + + +def write(output_dir, filename, rows): + """ + Write a list of dicts to a CSV file in output_dir. + Skips writing if rows is empty, but logs the skip. + """ + if not rows: + print(f" {filename}: no data, skipping") + return + + os.makedirs(output_dir, exist_ok=True) + path = os.path.join(output_dir, filename) + + with open(path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=rows[0].keys()) + writer.writeheader() + writer.writerows(rows) + + print(f" {filename}: {len(rows)} rows -> {path}") + + +def write_summary(output_dir, group, sections): + """ + Write a plain-text summary file listing section names and row counts. + sections: list of (label, row_count) tuples + """ + path = os.path.join(output_dir, "summary.txt") + lines = [ + "GitLab Audit Package", + f"Group: {group}", + "", + "Section Rows", + f"{'─' * 40}", + ] + for label, count in sections: + lines.append(f"{label:<35}{count}") + + with open(path, "w", encoding="utf-8") as f: + f.write("\n".join(lines) + "\n") + + print(f" summary.txt -> {path}") deleted file mode 100644 @@ -1,51 +0,0 @@ -""" -List all repositories (projects) for a user or organization in GitLab. -""" - -import requests - -BASE_URL = "https://gitlab.com/api/v4" -PRIVATE_TOKEN = "your_access_token" -USER_ID = "your_user_or_group_id" -TIMEOUT = 30 - -URL = f"{BASE_URL}/groups/{USER_ID}/projects" # Group URL -# URL = f"{BASE_URL}/users/{USER_ID}/projects" # User URL -HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN} - - -def list_projects(user_or_group_id): - PER_PAGE = 100 - page = 1 - projects = [] - - while True: - response = requests.get( - URL, - headers=HEADERS, - timeout=TIMEOUT, - params={"page": page, "per_page": PER_PAGE}, - ) - - if response.status_code == 200: - current_projects = response.json() - if not current_projects: - break - projects.extend(current_projects) - page += 1 - else: - print( - f"Failed to retrieve projects: {response.status_code} - {response.text}" - ) - break - - if projects: - print(f"Projects under ID: {user_or_group_id}:") - for project in projects: - print(f"- {project['name']} (ID: {project['id']})") - else: - print(f"No projects found for ID: {user_or_group_id}.") - - -if __name__ == "__main__": - list_projects(USER_ID) deleted file mode 100644 @@ -1,53 +0,0 @@ -""" -Gather all members of specified GitLab groups and projects and their access levels. - - Ref: https://docs.gitlab.com/api/members/ -""" - -import requests - -BASE_URL = "https://gitlab.com/api/v4" -PRIVATE_TOKEN = "your_access_token" -GROUP_IDS = ["group_id_1", "group_id_2"] # Add your group IDs here -PROJECT_IDS = ["project_id_1", "project_id_2"] # Add your project IDs here -TIMEOUT = 30 - -HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN} - - -def get_members(url, name): - response = requests.get(url, headers=HEADERS, timeout=TIMEOUT) - if response.status_code == 200: - members = response.json() - print(f"\n{name} Members:") - for member in members: - print( - f"Username: {member['username']}, Access Level: {member['access_level']}" - ) - else: - print( - f"Failed to fetch members for {name}: {response.status_code}, {response.text}" - ) - - -if __name__ == "__main__": - access_levels = """Access Level Roles: - 0 : No access - 5 : Minimal access - 10 : Guest - 15 : Planner - 20 : Reporter - 30 : Developer - 40 : Maintainer - 50 : Owner - 60 : Admin - """ - print(access_levels) - - for group_id in GROUP_IDS: - group_url = f"{BASE_URL}/groups/{group_id}/members" - get_members(group_url, f"Group {group_id}") - - for project_id in PROJECT_IDS: - project_url = f"{BASE_URL}/projects/{project_id}/members" - get_members(project_url, f"Project {project_id}") new file mode 100755 @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 +"""Launch the Audit Tools interactive terminal UI.""" + +from tui.app import main + +if __name__ == "__main__": + main() new file mode 100644 @@ -0,0 +1,10 @@ +"""Ensure the repository root is importable so tests can use absolute imports +(``from tui import ...``, ``from sampling.sampling_tool import ...``) regardless +of how pytest is invoked.""" + +import os +import sys + +ROOT = os.path.dirname(os.path.abspath(__file__)) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) @@ -4,6 +4,7 @@ xlrd PyYAML pytest requests +textual dash plotly urllib3>=2.7.0 new file mode 100644 @@ -0,0 +1,63 @@ +# Audit Tools — Interactive TUI + +A terminal UI that walks you through running an audit. It presents a platform +menu, collects connection details and check selection, then runs the existing +collectors with live progress. + +GitHub and GitLab are supported. Adding a platform is a matter of writing a +runner and a `Platform` descriptor in `tui/platforms.py` — the screens are +platform-agnostic. + +## Run it + +```bash +pip install -r requirements.txt +python audit_tui.py +``` + +The connection fields are pre-filled from environment variables if set: + +```bash +# GitHub +export GITHUB_ORG=my-org +export GITHUB_TOKEN=ghp_... # needs read:org and repo scopes + +# GitLab +export GITLAB_GROUP=my-group +export GITLAB_TOKEN=glpat-... # needs read_api scope +export GITLAB_URL=https://gitlab.example.com/api/v4 # self-hosted only +``` + +## Walkthrough + +1. **Platform** — choose GitHub or GitLab. +2. **Connection** — the audit subject (org / group), a masked token, and any + platform-specific fields (branch for GitHub; API base URL for GitLab). +3. **Checks** — toggle which checks to run. Plan-restricted checks (GitHub's + Enterprise audit log; GitLab's Premium and self-hosted checks) are off by + default. +4. **Run** — a progress bar and live log show each check completing with its row + count. Errors on a single check are reported without stopping the run. + +## Output + +The TUI writes the same package the platform's `audit.py` produces: +`<output>/github_audit_<org>_<date>/` or `<output>/gitlab_audit_<group>_<date>/`, +one CSV per check plus a `summary.txt`. It reuses each platform's collectors and +CSV reporter unchanged — the TUI is only an interactive driver around them. + +## Keys + +- `Esc` — back / return to menu +- `Ctrl+P` — command palette +- `q` — quit (from the menu) + +## Tests + +```bash +python -m pytest tui/tests +``` + +The tests stub the network-bound collectors, so they run offline: one suite +covers the run orchestration, another drives the app headlessly through every +screen. new file mode 100644 @@ -0,0 +1 @@ +"""Interactive terminal UI for running audit collectors.""" new file mode 100644 @@ -0,0 +1,337 @@ +""" +Audit Tools — interactive terminal UI. + +Presents a platform menu, walks the user through credentials and check +selection, then runs the selected platform's collectors with live progress. + +Run it with: + + python audit_tui.py +""" + +from typing import ClassVar + +from rich.text import Text +from textual import work +from textual.app import App, ComposeResult +from textual.containers import Center, Horizontal, Vertical +from textual.screen import Screen +from textual.widgets import ( + Button, + Footer, + Header, + Input, + Label, + ProgressBar, + RichLog, + SelectionList, + Static, +) +from textual.widgets.selection_list import Selection + +from tui import platforms +from tui.common import Check, ProgressEvent + + +class MenuScreen(Screen): + """Pick a platform to audit.""" + + BINDINGS: ClassVar[list] = [("q", "app.quit", "Quit")] + + def compose(self) -> ComposeResult: + yield Header() + with Center(), Vertical(id="menu-box"): + yield Static("Select a platform to audit", classes="prompt") + for platform in platforms.PLATFORMS: + label = platform.label + if not platform.enabled: + label = f"{label} — coming soon" + yield Button( + label, + id=platform.key, + variant="primary" if platform.enabled else "default", + disabled=not platform.enabled, + ) + yield Footer() + + def on_mount(self) -> None: + self.sub_title = "Select a platform" + + def on_button_pressed(self, event: Button.Pressed) -> None: + for platform in platforms.PLATFORMS: + if event.button.id == platform.key and platform.enabled: + self.app.platform = platform + self.app.push_screen(ConfigScreen()) + return + + +class ConfigScreen(Screen): + """Collect the connection details for the chosen platform.""" + + BINDINGS: ClassVar[list] = [("escape", "back", "Back")] + + def compose(self) -> ComposeResult: + platform = self.app.platform + yield Header() + with Center(), Vertical(id="form-box"): + yield Static( + f"{platform.label} audit — connection details", classes="prompt" + ) + for f in platform.fields: + yield Label(f.label) + yield Input( + value=platforms.prefill(f), + placeholder=f.placeholder, + password=f.password, + id=f.key, + ) + yield Static("", id="form-error", classes="error") + with Horizontal(classes="buttons"): + yield Button("Back", id="back") + yield Button("Continue", id="continue", variant="primary") + yield Footer() + + def on_mount(self) -> None: + platform = self.app.platform + self.sub_title = f"{platform.label} · connection" + self.query_one(f"#{platform.fields[0].key}", Input).focus() + + def action_back(self) -> None: + self.app.pop_screen() + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "back": + self.app.pop_screen() + elif event.button.id == "continue": + self._submit() + + def on_input_submitted(self, event: Input.Submitted) -> None: + self._submit() + + def _submit(self) -> None: + platform = self.app.platform + settings = {} + missing = [] + for f in platform.fields: + value = self.query_one(f"#{f.key}", Input).value.strip() + if not value: + value = f.default + if f.required and not value: + missing.append(f.label.lower()) + settings[f.key] = value + + if missing: + self.query_one("#form-error", Static).update( + f"Please provide: {', '.join(missing)}." + ) + return + + self.app.settings = settings + self.app.push_screen(ChecksScreen()) + + +class ChecksScreen(Screen): + """Choose which checks to run.""" + + BINDINGS: ClassVar[list] = [("escape", "back", "Back")] + + def compose(self) -> ComposeResult: + platform = self.app.platform + yield Header() + with Center(), Vertical(id="checks-box"): + yield Static("Select checks to run", classes="prompt") + yield SelectionList( + *[ + Selection( + self._prompt(c), + c.key, + c.key in platform.default_selection, + ) + for c in platform.checks + ], + id="checks", + ) + yield Static("", id="checks-error", classes="error") + with Horizontal(classes="buttons"): + yield Button("Back", id="back") + yield Button("Run audit", id="run", variant="primary") + yield Footer() + + def on_mount(self) -> None: + self.sub_title = f"{self.app.platform.label} · select checks" + self.query_one("#checks", SelectionList).focus() + + @staticmethod + def _prompt(check: Check) -> Text: + text = Text(check.label) + if check.note: + text.append(f" ({check.note})", style="dim italic") + return text + + def action_back(self) -> None: + self.app.pop_screen() + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "back": + self.app.pop_screen() + elif event.button.id == "run": + selected = list(self.query_one("#checks", SelectionList).selected) + if not selected: + self.query_one("#checks-error", Static).update( + "Select at least one check." + ) + return + self.app.selected_keys = selected + self.app.push_screen(RunScreen()) + + +class RunScreen(Screen): + """Run the selected checks with live progress.""" + + BINDINGS: ClassVar[list] = [("escape", "home", "Menu")] + + def compose(self) -> ComposeResult: + yield Header() + with Vertical(id="run-box"): + yield Static(id="run-target", classes="prompt") + yield ProgressBar(id="progress", show_eta=False) + yield RichLog(id="log", markup=True, highlight=False, wrap=True) + with Horizontal(classes="buttons"): + yield Button("Back to menu", id="menu", disabled=True) + yield Button("Quit", id="quit", disabled=True, variant="primary") + yield Footer() + + def on_mount(self) -> None: + platform = self.app.platform + settings = self.app.settings + keys = self.app.selected_keys + self.sub_title = f"{platform.label} · running" + self.output_dir = platform.output_dir(settings) + target = settings[platform.id_key] + self.query_one("#run-target", Static).update( + f"Auditing [b]{target}[/] · {len(keys)} checks · → {self.output_dir}" + ) + self.query_one("#progress", ProgressBar).update(total=len(keys), progress=0) + self.run_audit() + + @work(thread=True) + def run_audit(self) -> None: + platform = self.app.platform + settings = self.app.settings + keys = self.app.selected_keys + try: + platform.run( + settings, + self.output_dir, + keys, + lambda ev: self.app.call_from_thread(self._handle_event, ev), + ) + except Exception as e: # noqa: BLE001 - report unexpected failures in the UI + self.app.call_from_thread(self._log, f"[red]Run failed:[/] {e}") + finally: + self.app.call_from_thread(self._finish) + + def _log(self, markup: str) -> None: + self.query_one("#log", RichLog).write(markup) + + def _handle_event(self, ev: ProgressEvent) -> None: + if ev.kind == "fetch": + self._log(f"[dim]· {ev.label}…[/]") + elif ev.kind == "start": + self._log(f"[cyan]▶[/] {ev.label}…") + elif ev.kind == "done": + self._log(f"[green]✓[/] {ev.label} — [b]{ev.count}[/] rows") + self.query_one("#progress", ProgressBar).advance(1) + elif ev.kind == "error": + self._log(f"[red]✗[/] {ev.label} — {ev.message}") + self.query_one("#progress", ProgressBar).advance(1) + elif ev.kind == "summary": + self._log("") + self._log(f"[bold green]Done.[/] Package written to {ev.label}") + + def _finish(self) -> None: + self.query_one("#menu", Button).disabled = False + self.query_one("#quit", Button).disabled = False + + def action_home(self) -> None: + self.app.show_menu() + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "menu": + self.app.show_menu() + elif event.button.id == "quit": + self.app.exit() + + +class AuditApp(App): + TITLE = "Audit Tools" + + CSS = """ + Screen { + align: center middle; + } + #menu-box, #form-box, #checks-box { + width: 64; + height: auto; + padding: 1 2; + border: round $primary; + } + #run-box { + width: 90%; + height: 90%; + padding: 1 2; + border: round $primary; + } + .prompt { + text-style: bold; + margin-bottom: 1; + } + .error { + color: $error; + margin-top: 1; + } + Label { + margin-top: 1; + } + .buttons { + height: auto; + margin-top: 1; + align-horizontal: right; + } + .buttons Button { + margin-left: 2; + } + #menu-box Button { + width: 100%; + margin-top: 1; + } + #checks { + height: auto; + max-height: 14; + } + #log { + height: 1fr; + border: round $panel; + padding: 0 1; + margin-top: 1; + } + """ + + def on_mount(self) -> None: + self.platform = None + self.settings: dict = {} + self.selected_keys: list = [] + self.push_screen(MenuScreen()) + + def show_menu(self) -> None: + """Pop back to the platform menu.""" + while len(self.screen_stack) > 2: + self.pop_screen() + + +def main() -> None: + AuditApp().run() + + +if __name__ == "__main__": + main() new file mode 100644 @@ -0,0 +1,30 @@ +"""Shared types used by the platform runners and the TUI.""" + +from collections.abc import Callable +from dataclasses import dataclass + + +# ``arg`` describes how a collector is called: +# "base" -> fn(target, cfg) +# "collabs" -> fn(target, cfg, repo_collabs) (GitHub collaborator cache) +# "projects" -> fn(target, cfg, projects) (GitLab project cache) +@dataclass(frozen=True) +class Check: + key: str + label: str + fn: Callable + filename: str + arg: str = "base" + note: str = "" + + +# kind is one of: "fetch", "start", "done", "error", "summary". +@dataclass(frozen=True) +class ProgressEvent: + kind: str + label: str + count: int | None = None + message: str = "" + + +ProgressCallback = Callable[[ProgressEvent], None] new file mode 100644 @@ -0,0 +1,180 @@ +""" +Drive the existing GitHub audit collectors from the TUI. + +This module reuses the collectors and CSV reporter under +``applications/github`` unchanged. It exposes: + +- ``CHECKS``: the list of available audit checks the UI presents. +- ``run_audit``: run the selected checks, write the same output package + ``audit.py`` produces, and report progress through a callback. +""" + +import os +import sys +from collections.abc import Iterable +from datetime import date + +from tui.common import Check, ProgressCallback, ProgressEvent + +# Import the GitHub collectors as a namespaced package so the GitHub and GitLab +# collector packages (both named ``collectors`` on disk) can coexist in one +# process. Requires the repo root on sys.path. +_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _REPO_ROOT not in sys.path: + sys.path.insert(0, _REPO_ROOT) + +from applications.github.collectors import ( + audit_log, + branch_protections, + commits, + members, +) +from applications.github.reporters import csv_reporter + +# --- Check registry --------------------------------------------------------- + +# For GitHub, ``arg`` is "base" -> fn(org, cfg), "collabs" -> fn(org, cfg, +# repo_collabs), or "branch" -> fn(org, cfg, branch). + +CHECKS: list[Check] = [ + Check("member_roster", "Member roster", members.member_roster, "member_roster.csv"), + Check( + "two_factor", + "2FA disabled", + members.two_factor_disabled, + "two_factor_disabled.csv", + note="requires org owner token", + ), + Check( + "outside_collaborators", + "Outside collaborators", + members.outside_collaborators, + "outside_collaborators.csv", + arg="collabs", + ), + Check( + "privileged_access", + "Privileged access", + members.privileged_access, + "privileged_access.csv", + arg="collabs", + ), + Check( + "pending_invitations", + "Pending invitations", + members.pending_invitations, + "pending_invitations.csv", + ), + Check( + "team_permissions", + "Team permissions", + members.team_permissions, + "team_permissions.csv", + ), + Check( + "permission_matrix", + "Permission matrix", + members.permission_matrix, + "permission_matrix.csv", + arg="collabs", + ), + Check( + "branch_protections", + "Branch protections", + branch_protections.branch_protections, + "branch_protections.csv", + ), + Check("commits", "Commits", commits.commits, "commits.csv", arg="branch"), + Check( + "audit_log", + "Audit log (branch/ruleset changes)", + audit_log.audit_log, + "audit_log.csv", + note="requires GitHub Enterprise Cloud", + ), +] + +DEFAULT_SELECTION = [c.key for c in CHECKS if c.key != "audit_log"] + + +# --- Config + output helpers ------------------------------------------------ + + +def build_cfg(token: str) -> dict: + """Build the config dict the collectors expect (mirrors config.load()).""" + return { + "token": token, + "headers": { + "Authorization": f"token {token}", + "Accept": "application/vnd.github.v3+json", + }, + "timeout": 30, + } + + +def default_output_dir(out: str, org: str) -> str: + """Match the folder naming used by audit.py.""" + return os.path.join(out, f"github_audit_{org}_{date.today().isoformat()}") + + +# --- Runner ----------------------------------------------------------------- + + +def run_audit( + *, + org: str, + token: str, + output_dir: str, + branch: str, + selected_keys: Iterable[str], + on_event: ProgressCallback, +) -> list[tuple[str, int]]: + """ + Run the selected checks and write the audit package to ``output_dir``. + + A collector that raises is reported as an error and recorded with a count + of 0, matching audit.py's behavior of never aborting the whole run. + + Returns the list of (label, row_count) sections that was written to the + summary file. + """ + cfg = build_cfg(token) + selected = set(selected_keys) + checks = [c for c in CHECKS if c.key in selected] + + repo_collabs: list | None = None + if any(c.arg == "collabs" for c in checks): + on_event(ProgressEvent("fetch", "Repo collaborators (shared cache)")) + try: + repo_collabs = members.fetch_repo_collaborators(org, cfg) + except Exception as e: # noqa: BLE001 - surface, keep going + on_event( + ProgressEvent( + "error", "Repo collaborators (shared cache)", message=str(e) + ) + ) + repo_collabs = [] + + sections: list[tuple[str, int]] = [] + for c in checks: + on_event(ProgressEvent("start", c.label)) + try: + if c.arg == "collabs": + rows = c.fn(org, cfg, repo_collabs or []) + elif c.arg == "branch": + rows = c.fn(org, cfg, branch) + else: + rows = c.fn(org, cfg) + except Exception as e: # noqa: BLE001 - one bad check shouldn't kill the run + on_event(ProgressEvent("error", c.label, message=str(e))) + sections.append((c.label, 0)) + continue + + csv_reporter.write(output_dir, c.filename, rows) + sections.append((c.label, len(rows))) + on_event(ProgressEvent("done", c.label, count=len(rows))) + + csv_reporter.write_summary(output_dir, org, sections) + total = sum(n for _, n in sections) + on_event(ProgressEvent("summary", output_dir, count=total)) + return sections new file mode 100644 @@ -0,0 +1,163 @@ +""" +Drive the GitLab audit collectors from the TUI. + +Reuses the collectors and CSV reporter under ``applications/gitlab`` unchanged. +Mirrors github_runner: a ``CHECKS`` registry plus ``run_audit`` that writes the +same package ``applications/gitlab/audit.py`` produces and reports progress +through a callback. +""" + +import os +import sys +from collections.abc import Iterable +from datetime import date + +from tui.common import Check, ProgressCallback, ProgressEvent + +# Namespaced import so the GitHub and GitLab collector packages (both named +# ``collectors`` on disk) can coexist in one process. +_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _REPO_ROOT not in sys.path: + sys.path.insert(0, _REPO_ROOT) + +from applications.gitlab.collectors import ( + approvals, + audit_events, + branch_protections, + members, + pipelines, + projects, + settings, +) +from applications.gitlab.reporters import csv_reporter + +# --- Check registry --------------------------------------------------------- + +# For GitLab, ``arg`` is "base" -> fn(group, cfg) or "projects" -> fn(group, +# cfg, projects), where the project cache is fetched once and shared. + +CHECKS: list[Check] = [ + Check("group_members", "Group members", members.group_members, "group_members.csv"), + Check( + "projects", "Projects", projects.project_list, "projects.csv", arg="projects" + ), + Check( + "project_members", + "Project members", + members.project_members, + "project_members.csv", + arg="projects", + ), + Check( + "branch_protections", + "Branch protections", + branch_protections.branch_protections, + "branch_protections.csv", + arg="projects", + ), + Check( + "pipelines", + "Pipelines", + pipelines.pipelines, + "pipelines.csv", + arg="projects", + ), + Check( + "approval_rules", + "Approval rules", + approvals.approval_rules, + "approval_rules.csv", + arg="projects", + note="requires Premium/Ultimate", + ), + Check( + "audit_events", + "Audit events", + audit_events.audit_events, + "audit_events.csv", + note="requires Premium/Ultimate", + ), + Check( + "password_policy", + "Password policy", + settings.password_policy, + "password_policy.csv", + note="self-hosted, admin token", + ), +] + +_PREMIUM = {"approval_rules", "audit_events", "password_policy"} +DEFAULT_SELECTION = [c.key for c in CHECKS if c.key not in _PREMIUM] + + +# --- Config + output helpers ------------------------------------------------ + + +def build_cfg(token: str, base_url: str) -> dict: + """Build the config dict the collectors expect (mirrors config.load()).""" + return { + "token": token, + "base_url": base_url.rstrip("/"), + "headers": {"PRIVATE-TOKEN": token}, + "timeout": 30, + } + + +def default_output_dir(out: str, group: str) -> str: + """Match the folder naming used by applications/gitlab/audit.py.""" + safe_group = group.replace("/", "-") + return os.path.join(out, f"gitlab_audit_{safe_group}_{date.today().isoformat()}") + + +# --- Runner ----------------------------------------------------------------- + + +def run_audit( + *, + group: str, + token: str, + base_url: str, + output_dir: str, + selected_keys: Iterable[str], + on_event: ProgressCallback, +) -> list[tuple[str, int]]: + """ + Run the selected checks and write the audit package to ``output_dir``. + + A collector that raises is reported as an error and recorded with a count + of 0, so one bad check never aborts the whole run. + """ + cfg = build_cfg(token, base_url) + selected = set(selected_keys) + checks = [c for c in CHECKS if c.key in selected] + + project_cache: list | None = None + if any(c.arg == "projects" for c in checks): + on_event(ProgressEvent("fetch", "Projects (shared cache)")) + try: + project_cache = projects.fetch_projects(group, cfg) + except Exception as e: # noqa: BLE001 - surface, keep going + on_event(ProgressEvent("error", "Projects (shared cache)", message=str(e))) + project_cache = [] + + sections: list[tuple[str, int]] = [] + for c in checks: + on_event(ProgressEvent("start", c.label)) + try: + if c.arg == "projects": + rows = c.fn(group, cfg, project_cache or []) + else: + rows = c.fn(group, cfg) + except Exception as e: # noqa: BLE001 - one bad check shouldn't kill the run + on_event(ProgressEvent("error", c.label, message=str(e))) + sections.append((c.label, 0)) + continue + + csv_reporter.write(output_dir, c.filename, rows) + sections.append((c.label, len(rows))) + on_event(ProgressEvent("done", c.label, count=len(rows))) + + csv_reporter.write_summary(output_dir, group, sections) + total = sum(n for _, n in sections) + on_event(ProgressEvent("summary", output_dir, count=total)) + return sections new file mode 100644 @@ -0,0 +1,144 @@ +""" +Platform descriptors that let the TUI drive any collector runner. + +Each Platform declares its connection form (``fields``), its checks, and how to +compute the output directory and run the audit. The screens in app.py are +written against this interface, so adding a platform is data, not new UI. +""" + +import os +from collections.abc import Callable +from dataclasses import dataclass, field + +from tui import github_runner, gitlab_runner +from tui.common import Check + + +@dataclass(frozen=True) +class Field: + """One input on the connection screen.""" + + key: str + label: str + placeholder: str = "" + default: str = "" + password: bool = False + required: bool = False + env: str | None = None # environment variable used to pre-fill the value + + +@dataclass(frozen=True) +class Platform: + key: str + label: str + id_key: str # which field is the audit subject (org / group) + fields: list[Field] + checks: list[Check] + default_selection: list[str] + output_dir: Callable[[dict], str] # (settings) -> path + run: Callable[..., object] # (settings, output_dir, selected_keys, on_event) + enabled: bool = True + note: str = field(default="") + + +def _prefill(f: Field) -> str: + if f.env: + value = os.environ.get(f.env, "").strip() + if value: + return value + return f.default + + +def _github_output_dir(s: dict) -> str: + return github_runner.default_output_dir(s["out"], s["org"]) + + +def _github_run(s: dict, output_dir, selected_keys, on_event): + return github_runner.run_audit( + org=s["org"], + token=s["token"], + output_dir=output_dir, + branch=s["branch"], + selected_keys=selected_keys, + on_event=on_event, + ) + + +def _gitlab_output_dir(s: dict) -> str: + return gitlab_runner.default_output_dir(s["out"], s["group"]) + + +def _gitlab_run(s: dict, output_dir, selected_keys, on_event): + return gitlab_runner.run_audit( + group=s["group"], + token=s["token"], + base_url=s["base_url"], + output_dir=output_dir, + selected_keys=selected_keys, + on_event=on_event, + ) + + +GITHUB = Platform( + key="github", + label="GitHub", + id_key="org", + fields=[ + Field("org", "Organization", "my-org", required=True, env="GITHUB_ORG"), + Field( + "token", + "Personal access token", + "ghp_… (read:org, repo)", + password=True, + required=True, + env="GITHUB_TOKEN", + ), + Field("out", "Output directory", default="./output"), + Field("branch", "Branch (for commit history)", default="main"), + ], + checks=github_runner.CHECKS, + default_selection=github_runner.DEFAULT_SELECTION, + output_dir=_github_output_dir, + run=_github_run, +) + +GITLAB = Platform( + key="gitlab", + label="GitLab", + id_key="group", + fields=[ + Field( + "group", + "Group ID or path", + "e.g. 1234567 or my-group", + required=True, + env="GITLAB_GROUP", + ), + Field( + "token", + "Personal access token", + "glpat-… (read_api)", + password=True, + required=True, + env="GITLAB_TOKEN", + ), + Field( + "base_url", + "API base URL (self-hosted)", + default="https://gitlab.com/api/v4", + env="GITLAB_URL", + ), + Field("out", "Output directory", default="./output"), + ], + checks=gitlab_runner.CHECKS, + default_selection=gitlab_runner.DEFAULT_SELECTION, + output_dir=_gitlab_output_dir, + run=_gitlab_run, +) + +PLATFORMS = [GITHUB, GITLAB] + + +def prefill(f: Field) -> str: + """Public accessor for a field's pre-filled value (env var or default).""" + return _prefill(f) new file mode 100644 @@ -0,0 +1,127 @@ +"""Headless smoke tests for the Textual app. + +Drives the app through its screens with a Pilot, stubbing the network-bound +run_audit so no real GitHub calls are made. Uses asyncio.run so the suite does +not require the pytest-asyncio plugin. +""" + +import asyncio + +from textual.widgets import Button, Input + +from tui import github_runner as gh +from tui.app import AuditApp, ChecksScreen, ConfigScreen, MenuScreen, RunScreen + + +def _run(coro): + asyncio.run(coro) + + +def test_full_navigation(monkeypatch): + def fake_run_audit(*, org, token, output_dir, branch, selected_keys, on_event): + on_event(gh.ProgressEvent("start", "Member roster")) + on_event(gh.ProgressEvent("done", "Member roster", count=3)) + on_event(gh.ProgressEvent("summary", output_dir, count=3)) + return [("Member roster", 3)] + + monkeypatch.setattr("tui.github_runner.run_audit", fake_run_audit) + + async def scenario(): + app = AuditApp() + async with app.run_test(size=(120, 40)) as pilot: + await pilot.pause() + assert isinstance(app.screen, MenuScreen) + + await pilot.click("#github") + await pilot.pause() + assert isinstance(app.screen, ConfigScreen) + + app.screen.query_one("#org", Input).value = "acme" + app.screen.query_one("#token", Input).value = "tok" + await pilot.click("#continue") + await pilot.pause() + assert isinstance(app.screen, ChecksScreen) + assert app.settings["org"] == "acme" + + await pilot.click("#run") + await pilot.pause() + assert isinstance(app.screen, RunScreen) + + await app.workers.wait_for_complete() + await pilot.pause() + + # When the run finishes, the exit buttons become enabled. + assert app.screen.query_one("#menu", Button).disabled is False + assert app.screen.query_one("#quit", Button).disabled is False + + _run(scenario()) + + +def test_config_requires_org_and_token(monkeypatch): + # Make sure env vars don't pre-fill the fields for this test. + monkeypatch.delenv("GITHUB_ORG", raising=False) + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + + async def scenario(): + app = AuditApp() + async with app.run_test(size=(120, 40)) as pilot: + await pilot.click("#github") + await pilot.pause() + # Continue with empty fields -> stays on ConfigScreen with an error. + await pilot.click("#continue") + await pilot.pause() + assert isinstance(app.screen, ConfigScreen) + error_text = str(app.screen.query_one("#form-error").render()) + assert "provide" in error_text.lower() + + _run(scenario()) + + +def test_gitlab_is_enabled(): + async def scenario(): + app = AuditApp() + async with app.run_test(size=(120, 40)) as pilot: + await pilot.pause() + assert app.screen.query_one("#gitlab", Button).disabled is False + + _run(scenario()) + + +def test_gitlab_navigation(monkeypatch): + monkeypatch.delenv("GITLAB_GROUP", raising=False) + monkeypatch.delenv("GITLAB_TOKEN", raising=False) + + def fake_run_audit(*, group, token, base_url, output_dir, selected_keys, on_event): + on_event(gh.ProgressEvent("done", "Group members", count=7)) + on_event(gh.ProgressEvent("summary", output_dir, count=7)) + return [("Group members", 7)] + + monkeypatch.setattr("tui.gitlab_runner.run_audit", fake_run_audit) + + async def scenario(): + app = AuditApp() + async with app.run_test(size=(120, 40)) as pilot: + await pilot.pause() + await pilot.click("#gitlab") + await pilot.pause() + assert isinstance(app.screen, ConfigScreen) + + app.screen.query_one("#group", Input).value = "my-group" + app.screen.query_one("#token", Input).value = "glpat-x" + await pilot.click("#continue") + await pilot.pause() + assert isinstance(app.screen, ChecksScreen) + assert app.settings["group"] == "my-group" + # Self-hosted URL defaults to gitlab.com. + assert app.settings["base_url"] == "https://gitlab.com/api/v4" + + await pilot.click("#run") + await pilot.pause() + assert isinstance(app.screen, RunScreen) + assert "gitlab_audit_my-group" in app.screen.output_dir + + await app.workers.wait_for_complete() + await pilot.pause() + assert app.screen.query_one("#menu", Button).disabled is False + + _run(scenario()) new file mode 100644 @@ -0,0 +1,114 @@ +"""Tests for the TUI's GitHub audit orchestration. + +These stub out the network-bound collectors and verify run_audit's wiring: +argument dispatch, the shared collaborator cache, per-check error handling, +and that the CSV package (per-check files + summary) is written. +""" + +import csv +import os + +import pytest + +from tui import github_runner as r + + +@pytest.fixture +def fake_checks(monkeypatch): + """Replace the real registry with stubbed collectors and record calls.""" + calls = {} + + def base_fn(org, cfg): + calls["base"] = (org, cfg) + return [{"login": "alice"}, {"login": "bob"}] + + def collabs_fn(org, cfg, repo_collabs): + calls["collabs"] = (org, cfg, repo_collabs) + return [{"repo": e["repo"]} for e in repo_collabs] + + def branch_fn(org, cfg, branch): + calls["branch"] = (org, cfg, branch) + return [{"branch": branch}] + + def boom_fn(org, cfg): + raise RuntimeError("kaboom") + + checks = [ + r.Check("base", "Base", base_fn, "base.csv"), + r.Check("collabs", "Collabs", collabs_fn, "collabs.csv", arg="collabs"), + r.Check("branch", "Branch", branch_fn, "branch.csv", arg="branch"), + r.Check("boom", "Boom", boom_fn, "boom.csv"), + ] + monkeypatch.setattr(r, "CHECKS", checks) + + fetch_calls = [] + + def fake_fetch(org, cfg): + fetch_calls.append(org) + return [{"repo": "repo1", "collaborators": []}] + + monkeypatch.setattr(r.members, "fetch_repo_collaborators", fake_fetch) + + return calls, fetch_calls + + +def run(tmp_path, keys, fake_checks, branch="main"): + events = [] + sections = r.run_audit( + org="acme", + token="tok", + output_dir=str(tmp_path), + branch=branch, + selected_keys=keys, + on_event=events.append, + ) + return events, sections + + +def test_argument_dispatch_and_files(tmp_path, fake_checks): + calls, _ = fake_checks + run(tmp_path, ["base", "collabs", "branch"], fake_checks, "dev") + + # Each collector was called with the right signature. + assert calls["base"][0] == "acme" + assert calls["collabs"][2] == [{"repo": "repo1", "collaborators": []}] + assert calls["branch"][2] == "dev" + + # CSV files were written for each check, plus the summary. + for name in ("base.csv", "collabs.csv", "branch.csv", "summary.txt"): + assert os.path.exists(tmp_path / name), name + + with open(tmp_path / "base.csv", newline="") as f: + assert len(list(csv.DictReader(f))) == 2 + + +def test_collab_cache_fetched_once(tmp_path, fake_checks): + _, fetch_calls = fake_checks + run(tmp_path, ["collabs", "base"], fake_checks) + assert fetch_calls == ["acme"] # fetched exactly once + + +def test_collab_cache_skipped_when_not_needed(tmp_path, fake_checks): + _, fetch_calls = fake_checks + run(tmp_path, ["base"], fake_checks) + assert fetch_calls == [] # no collabs check selected -> no fetch + + +def test_failing_check_does_not_abort_run(tmp_path, fake_checks): + events, sections = run(tmp_path, ["boom", "base"], fake_checks) + + kinds = [(e.kind, e.label) for e in events] + assert ("error", "Boom") in kinds + assert ("done", "Base") in kinds # base still ran after boom failed + + labels = dict(sections) + assert labels["Boom"] == 0 + assert labels["Base"] == 2 + + +def test_summary_event_totals_rows(tmp_path, fake_checks): + events, _ = run(tmp_path, ["base", "branch"], fake_checks) + summary = [e for e in events if e.kind == "summary"] + assert len(summary) == 1 + assert summary[0].count == 3 # 2 base + 1 branch + assert summary[0].label == str(tmp_path) new file mode 100644 @@ -0,0 +1,113 @@ +"""Tests for the TUI's GitLab audit orchestration. + +Stub the network-bound collectors and verify run_audit's wiring: base vs +project-scoped dispatch, the shared project cache, per-check error handling, +and that the CSV package (per-check files + summary) is written. +""" + +import csv +import os + +import pytest + +from tui import gitlab_runner as r + + +@pytest.fixture +def fake_checks(monkeypatch): + calls = {} + + def base_fn(group, cfg): + calls["base"] = (group, cfg) + return [{"username": "alice"}, {"username": "bob"}] + + def projects_fn(group, cfg, projects): + calls["projects"] = (group, cfg, projects) + return [{"project": p["path_with_namespace"]} for p in projects] + + def boom_fn(group, cfg): + raise RuntimeError("kaboom") + + checks = [ + r.Check("base", "Base", base_fn, "base.csv"), + r.Check("projects", "Projects", projects_fn, "projects.csv", arg="projects"), + r.Check("boom", "Boom", boom_fn, "boom.csv"), + ] + monkeypatch.setattr(r, "CHECKS", checks) + + fetch_calls = [] + + def fake_fetch(group, cfg): + fetch_calls.append(group) + return [{"id": 1, "path_with_namespace": "grp/proj"}] + + monkeypatch.setattr(r.projects, "fetch_projects", fake_fetch) + + return calls, fetch_calls + + +def run(tmp_path, keys, base_url="https://gitlab.com/api/v4"): + events = [] + sections = r.run_audit( + group="grp", + token="tok", + base_url=base_url, + output_dir=str(tmp_path), + selected_keys=keys, + on_event=events.append, + ) + return events, sections + + +def test_argument_dispatch_and_files(tmp_path, fake_checks): + calls, _ = fake_checks + run(tmp_path, ["base", "projects"]) + + assert calls["base"][0] == "grp" + assert calls["projects"][2] == [{"id": 1, "path_with_namespace": "grp/proj"}] + + for name in ("base.csv", "projects.csv", "summary.txt"): + assert os.path.exists(tmp_path / name), name + + with open(tmp_path / "projects.csv", newline="") as f: + rows = list(csv.DictReader(f)) + assert rows == [{"project": "grp/proj"}] + + +def test_project_cache_fetched_once(tmp_path, fake_checks): + _, fetch_calls = fake_checks + run(tmp_path, ["projects", "base"]) + assert fetch_calls == ["grp"] + + +def test_project_cache_skipped_when_not_needed(tmp_path, fake_checks): + _, fetch_calls = fake_checks + run(tmp_path, ["base"]) + assert fetch_calls == [] + + +def test_failing_check_does_not_abort_run(tmp_path, fake_checks): + events, sections = run(tmp_path, ["boom", "base"]) + + kinds = [(e.kind, e.label) for e in events] + assert ("error", "Boom") in kinds + assert ("done", "Base") in kinds + + labels = dict(sections) + assert labels["Boom"] == 0 + assert labels["Base"] == 2 + + +def test_base_url_reaches_cfg(tmp_path, fake_checks): + calls, _ = fake_checks + run(tmp_path, ["base"], base_url="https://gitlab.example.com/api/v4/") + # build_cfg strips the trailing slash. + assert calls["base"][1]["base_url"] == "https://gitlab.example.com/api/v4" + assert calls["base"][1]["headers"]["PRIVATE-TOKEN"] == "tok" + + +def test_premium_checks_off_by_default(): + assert "approval_rules" not in r.DEFAULT_SELECTION + assert "audit_events" not in r.DEFAULT_SELECTION + assert "password_policy" not in r.DEFAULT_SELECTION + assert "group_members" in r.DEFAULT_SELECTION