31f6eb371d · audit-labs/audit-tools

README.org +12 −1

 |----------------------+------------------------------------------------------------------------------|
 | =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                                                         |
 | =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
 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:

applications/__init__.py added

applications/github/__init__.py added

applications/gitlab/README.md +42 −144

	@@ -1,160 +1,58 @@
1	# `approvals.py`	1	> NOTE: The token used across all collectors needs at least the `read_api`
		2	> scope. Some checks need more:
		3	> - Approval rules and audit events require a GitLab Premium or Ultimate
		4	> subscription.
		5	> - Password policy reads instance application settings, which require an
		6	> admin token on a self-hosted instance (not available on GitLab.com).
		7	>
		8	> Checks that are unavailable are skipped with a warning; the rest still run.
2		9
3	\\This script requires an active Premium or Ultimate subscription.\*\\	10	---
4		11
5	``` bash	12	# `audit.py` — Unified GitLab Audit Tool
6	python ./approvals.py
7	```
8		13
9	``` text	14	Runs all collectors against a GitLab group (including its subgroups) and writes
10	Rule: All Members	15	a timestamped audit package to disk.
11	Approvals Required: 1
12	Rule type: any_approver
13	Rule: Default
14	Approvals Required: 1
15	Rule type: regular
16	Protected Branch: master
17	Eligible Approver: Christian Cleberg
18	```
19		16
20	# `branch_protections.py`	17	## Setup
21		18
22	``` bash	19	```bash
23	python ./branch_protections.py	20	export GITLAB_TOKEN=your_token
		21	export GITLAB_GROUP=your_group_id_or_path
		22	# Self-hosted only:
		23	export GITLAB_URL=https://gitlab.example.com/api/v4
24	```	24	```
25		25
26	``` json	26	## Usage
27	[
28	{
29	"id": 148448212,
30	"name": "main",
31	"push_access_levels": [
32	{
33	"id": 185900194,
34	"access_level": 40,
35	"access_level_description": "Maintainers",
36	"deploy_key_id": null,
37	"user_id": null,
38	"group_id": null
39	}
40	],
41	"merge_access_levels": [
42	{
43	"id": 156461000,
44	"access_level": 40,
45	"access_level_description": "Maintainers",
46	"user_id": null,
47	"group_id": null
48	}
49	],
50	"allow_force_push": false,
51	"unprotect_access_levels": [],
52	"code_owner_approval_required": false,
53	"inherited": false
54	}
55	]
56	```
57		27
58	# `passwords.py`	28	```bash
		29	# Basic run — uses GITLAB_TOKEN and GITLAB_GROUP from environment
		30	python audit.py
59		31
60	**This script does not apply to GitLab.com. This is for self-hosted	32	# Override group, set output directory
61	instances only.**	33	python audit.py --group my-group --out ./output
62		34
63	``` bash	35	# Point at a self-hosted instance
64	python ./passwords.py	36	python audit.py --url https://gitlab.example.com/api/v4
65	```	37	```
66		38
67	``` text	39	The group may be a numeric ID (`1234567`) or a URL path (`my-group/sub-group`).
68	# TODO: Need access to a self-hosted version of GitLab to test this out.
69	```
70		40
71	# `pipelines.py`	41	## Output
72		42
73	``` bash	43	Creates a directory: `<out>/gitlab_audit_<group>_<YYYY-MM-DD>/`
74	python ./pipelines.py
75	```
76
77	``` text
78	Pipeline ID: 1754222228
79	Status: failed
80	Ref: master
81	Created At: 2025-04-06T03:39:15.065Z
82	Duration: N/A seconds
83	Configuration: N/A
84	Pipeline ID: 1754221831
85	Status: failed
86	Ref: pr-1
87	Created At: 2025-04-06T03:37:42.333Z
88	Duration: N/A seconds
89	Configuration: N/A
90	Pipeline ID: 1754220271
91	Status: failed
92	Ref: pr-1
93	Created At: 2025-04-06T03:33:38.606Z
94	Duration: N/A seconds
95	Configuration: N/A
96	Pipeline ID: 1754214637
97	Status: failed
98	Ref: master
99	Created At: 2025-04-06T03:21:39.902Z
100	Duration: N/A seconds
101	Configuration: N/A
102	```
103
104	# `provisioning.py`
105
106	\\This script requires an active Premium or Ultimate subscription.\*\\
107
108	``` bash
109	python ./provisioning.py
110	```
111
112	``` text
113	Group: 105300140
114	2025-04-08T03:33:17.055Z : Action: member_created, Member: 128029250, Author: 24608590
115	```
116		44
117	# `repositories.py`	45	\| File \| Contents \|
		46	\|---\|---\|
		47	\| `group_members.csv` \| Group members with access level and role \|
		48	\| `projects.csv` \| All projects in the group and subgroups \|
		49	\| `project_members.csv` \| Members and access levels for every project \|
		50	\| `branch_protections.csv` \| Protected-branch settings across all projects \|
		51	\| `pipelines.csv` \| CI/CD pipeline history across all projects \|
		52	\| `approval_rules.csv` \| Merge-request approval rules (Premium/Ultimate) \|
		53	\| `audit_events.csv` \| Group membership audit events (Premium/Ultimate) \|
		54	\| `password_policy.csv` \| Instance password policy (self-hosted, admin token) \|
		55	\| `summary.txt` \| Row counts per section \|
118		56
119	``` shell	57	The per-project checks reuse a single enumeration of the group's projects, so
120	python ./repositories.py	58	the group is listed only once per run.
121	```
122
123	``` text
124	# User ID Example
125	Projects under ID: ccleberg:
126	- audit-tools (ID: 68757698)
127	- cleberg.net (ID: 68701468)
128
129	# Group ID Example
130	Projects under ID: phryq:
131	- Yoshi Cli (ID: 68757750)
132	- pages-demo (ID: 68757186)
133	```
134
135	# `users.py`
136
137	``` bash
138	python ./users.py
139	```
140
141	``` text
142	Access Level Roles:
143	0 : No access
144	5 : Minimal access
145	10 : Guest
146	15 : Planner
147	20 : Reporter
148	30 : Developer
149	40 : Maintainer
150	50 : Owner
151	60 : Admin
152
153
154	Group 97083755 Members:
155	Username: ccleberg, Access Level: 50
156
157	Project 68701468 Members:
158	Username: ccleberg, Access Level: 50
159	Username: project_68701468_bot_2c7ee010a479c0e48cdb4c7c5cfae886, Access Level: 40
160	```

applications/gitlab/__init__.py added

applications/gitlab/approvals.py deleted −38

	@@ -1,38 +0,0 @@
1	"""
2	Extract merge request approval rules and their statuses in GitLab.
3	"""
4
5	import requests
6
7	BASE_URL = "https://gitlab.com/api/v4"
8	PRIVATE_TOKEN = "your_access_token"
9	PROJECT_ID = "your_project_id"
10	TIMEOUT = 30
11
12	URL = f"{BASE_URL}/projects/{PROJECT_ID}/approval_rules"
13	HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN}
14
15	if __name__ == "__main__":
16	# Get approval rules
17	response = requests.get(URL, headers=HEADERS, timeout=TIMEOUT)
18	if response.status_code == 200:
19	approval_rules = response.json()
20	for rule in approval_rules:
21	name = rule["name"]
22	approvals_required = rule["approvals_required"]
23	rule_type = rule["rule_type"]
24	protected_branches = rule["protected_branches"]
25	eligible_approvers = rule["eligible_approvers"]
26	print(f"Rule: {name}")
27	print(f" Approvals Required: {approvals_required}")
28	print(f" Rule type: {rule_type}")
29	for branch in protected_branches:
30	branch_name = branch["name"]
31	print(f" Protected Branch: {branch_name}")
32	for approver in eligible_approvers:
33	approver_username = approver["name"]
34	print(f" Eligible Approver: {approver_username}")
35	else:
36	print(
37	f"Failed to fetch approval rules: {response.status_code}, {response.text}"
38	)

applications/gitlab/audit.py added +149

@@ -0,0 +1,149 @@
	1	"""
	2	GitLab audit CLI.
	3
	4	Runs all collectors against a GitLab group and writes a timestamped audit
	5	package to an output directory.
	6
	7	Usage:
	8	export GITLAB_TOKEN=your_token
	9	export GITLAB_GROUP=your_group_id_or_path
	10
	11	python audit.py
	12	python audit.py --group my-group
	13	python audit.py --group my-group --out ./output
	14	python audit.py --group my-group --url https://gitlab.example.com/api/v4
	15
	16	Output:
	17	<out>/gitlab_audit_<group>_<date>/
	18	group_members.csv
	19	projects.csv
	20	project_members.csv
	21	branch_protections.csv
	22	pipelines.csv
	23	approval_rules.csv
	24	audit_events.csv
	25	password_policy.csv
	26	summary.txt
	27	"""
	28
	29	import argparse
	30	import os
	31	import sys
	32	from datetime import date
	33
	34	import config
	35	from collectors import (
	36	approvals,
	37	audit_events,
	38	branch_protections,
	39	members,
	40	pipelines,
	41	projects,
	42	settings,
	43	)
	44	from reporters import csv_reporter
	45
	46
	47	def parse_args():
	48	parser = argparse.ArgumentParser(
	49	description="Generate a GitLab audit package for a group."
	50	)
	51	parser.add_argument(
	52	"--group",
	53	help="GitLab group ID or path. Overrides GITLAB_GROUP env var.",
	54	)
	55	parser.add_argument(
	56	"--url",
	57	help="GitLab API base URL. Overrides GITLAB_URL env var. "
	58	"Default: https://gitlab.com/api/v4",
	59	)
	60	parser.add_argument(
	61	"--out",
	62	default="./output",
	63	help="Directory to write the audit package into. Default: ./output",
	64	)
	65	return parser.parse_args()
	66
	67
	68	def run():
	69	args = parse_args()
	70	cfg = config.load(group_override=args.group, base_url_override=args.url)
	71	group = cfg["group"]
	72
	73	safe_group = group.replace("/", "-")
	74	output_dir = os.path.join(
	75	args.out, f"gitlab_audit_{safe_group}_{date.today().isoformat()}"
	76	)
	77
	78	print(f"GitLab Audit — {group}")
	79	print(f"Output directory: {output_dir}")
	80	print()
	81
	82	sections = []
	83
	84	def collect(label, fn, filename, *fn_args):
	85	print(f"Collecting: {label}...")
	86	try:
	87	rows = fn(*fn_args)
	88	except Exception as e:
	89	print(f" Error: {e}", file=sys.stderr)
	90	rows = []
	91	csv_reporter.write(output_dir, filename, rows)
	92	sections.append((label, len(rows)))
	93	return rows
	94
	95	print("Enumerating projects (shared cache)...")
	96	try:
	97	project_cache = projects.fetch_projects(group, cfg)
	98	except Exception as e:
	99	print(f" Error enumerating projects: {e}", file=sys.stderr)
	100	project_cache = []
	101
	102	collect("Group members", members.group_members, "group_members.csv", group, cfg)
	103	collect(
	104	"Projects", projects.project_list, "projects.csv", group, cfg, project_cache
	105	)
	106	collect(
	107	"Project members",
	108	members.project_members,
	109	"project_members.csv",
	110	group,
	111	cfg,
	112	project_cache,
	113	)
	114	collect(
	115	"Branch protections",
	116	branch_protections.branch_protections,
	117	"branch_protections.csv",
	118	group,
	119	cfg,
	120	project_cache,
	121	)
	122	collect(
	123	"Pipelines", pipelines.pipelines, "pipelines.csv", group, cfg, project_cache
	124	)
	125	collect(
	126	"Approval rules",
	127	approvals.approval_rules,
	128	"approval_rules.csv",
	129	group,
	130	cfg,
	131	project_cache,
	132	)
	133	collect("Audit events", audit_events.audit_events, "audit_events.csv", group, cfg)
	134	collect(
	135	"Password policy",
	136	settings.password_policy,
	137	"password_policy.csv",
	138	group,
	139	cfg,
	140	)
	141
	142	print()
	143	csv_reporter.write_summary(output_dir, group, sections)
	144	print()
	145	print("Done.")
	146
	147
	148	if __name__ == "__main__":
	149	run()

applications/gitlab/branch_protections.py deleted −25

	@@ -1,25 +0,0 @@
1	"""
2	List all branch protection rules and their configurations in GitLab.
3	"""
4
5	import requests
6	import json
7
8	BASE_URL = "https://gitlab.com/api/v4"
9	PRIVATE_TOKEN = "your_access_token"
10	PROJECT_ID = "your_project_id"
11	TIMEOUT = 30
12
13	URL = f"{BASE_URL}/projects/{PROJECT_ID}/protected_branches"
14	HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN}
15
16	if __name__ == "__main__":
17	# Get protected branches
18	response = requests.get(URL, headers=HEADERS, timeout=TIMEOUT)
19	if response.status_code == 200:
20	protected_branches = response.json()
21	print(json.dumps(protected_branches, indent=4))
22	else:
23	print(
24	f"Failed to fetch protected branches: {response.status_code}, {response.text}"
25	)

applications/gitlab/collectors/__init__.py added

applications/gitlab/collectors/api.py added +41

@@ -0,0 +1,41 @@
	1	"""Shared GitLab API helpers."""
	2
	3	from urllib.parse import quote
	4
	5	import requests
	6
	7	DEFAULT_BASE_URL = "https://gitlab.com/api/v4"
	8
	9
	10	def enc(value):
	11	"""URL-encode a group or project identifier.
	12
	13	GitLab accepts either a numeric ID or a URL-encoded path (e.g.
	14	``my-group/sub-group``). Numeric IDs pass through unchanged.
	15	"""
	16	return quote(str(value), safe="")
	17
	18
	19	def paginate(url, cfg, params=None):
	20	"""Fetch all pages from a GitLab endpoint using the X-Next-Page header."""
	21	results = []
	22	p = dict(params or {})
	23	p["per_page"] = 100
	24	page = 1
	25
	26	while True:
	27	p["page"] = page
	28	resp = requests.get(
	29	url, headers=cfg["headers"], params=p, timeout=cfg["timeout"]
	30	)
	31	resp.raise_for_status()
	32	data = resp.json()
	33	if not data:
	34	break
	35	results.extend(data)
	36	next_page = resp.headers.get("X-Next-Page")
	37	if not next_page:
	38	break
	39	page = int(next_page)
	40
	41	return results

applications/gitlab/collectors/approvals.py added +48

@@ -0,0 +1,48 @@
	1	"""
	2	Collect merge-request approval rules for every project in the group.
	3
	4	Approval rules require a GitLab Premium or Ultimate subscription. Projects that
	5	return 403/404 (feature unavailable) are skipped with a warning.
	6	"""
	7
	8	import sys
	9
	10	import requests
	11
	12	from .api import paginate
	13
	14
	15	def approval_rules(group, cfg, projects):
	16	rows = []
	17	for p in projects:
	18	try:
	19	rules = paginate(
	20	f"{cfg['base_url']}/projects/{p['id']}/approval_rules", cfg
	21	)
	22	except requests.HTTPError as e:
	23	if e.response is not None and e.response.status_code in (403, 404):
	24	print(
	25	f" Skipping {p.get('path_with_namespace', p['id'])}: "
	26	f"approval_rules returned {e.response.status_code}",
	27	file=sys.stderr,
	28	)
	29	continue
	30	raise
	31	for rule in rules:
	32	approvers = ", ".join(
	33	a.get("name", "") for a in rule.get("eligible_approvers", [])
	34	)
	35	branches = ", ".join(
	36	b.get("name", "") for b in rule.get("protected_branches", [])
	37	)
	38	rows.append(
	39	{
	40	"project": p.get("path_with_namespace", ""),
	41	"rule": rule.get("name", ""),
	42	"rule_type": rule.get("rule_type", ""),
	43	"approvals_required": rule.get("approvals_required", 0),
	44	"protected_branches": branches or "(all)",
	45	"eligible_approvers": approvers or "(none)",
	46	}
	47	)
	48	return rows

applications/gitlab/collectors/audit_events.py added +46

@@ -0,0 +1,46 @@
	1	"""
	2	Collect group membership audit events (created / updated / destroyed).
	3
	4	Group audit events require a GitLab Premium or Ultimate subscription. Returns an
	5	empty list with a warning if the endpoint is unavailable (403/404).
	6	"""
	7
	8	import sys
	9
	10	import requests
	11
	12	from .api import enc, paginate
	13
	14	MEMBER_ACTIONS = {"member_created", "member_updated", "member_destroyed"}
	15
	16
	17	def audit_events(group, cfg):
	18	try:
	19	events = paginate(f"{cfg['base_url']}/groups/{enc(group)}/audit_events", cfg)
	20	except requests.HTTPError as e:
	21	if e.response is not None and e.response.status_code in (403, 404):
	22	print(
	23	"Warning: group audit events require GitLab Premium/Ultimate and "
	24	"owner access -- skipping.",
	25	file=sys.stderr,
	26	)
	27	return []
	28	raise
	29
	30	rows = []
	31	for event in events:
	32	action = event.get("event_name", "")
	33	if action not in MEMBER_ACTIONS:
	34	continue
	35	details = event.get("details", {})
	36	rows.append(
	37	{
	38	"created_at": event.get("created_at", ""),
	39	"action": action,
	40	"member_id": details.get("member_id", ""),
	41	"target": details.get("target_details", ""),
	42	"author_id": event.get("author_id", ""),
	43	"entity_type": event.get("entity_type", ""),
	44	}
	45	)
	46	return rows

applications/gitlab/collectors/branch_protections.py added +44

@@ -0,0 +1,44 @@
	1	"""Collect protected-branch settings for every project in the group."""
	2
	3	import sys
	4
	5	import requests
	6
	7	from .api import paginate
	8
	9
	10	def _levels(entries):
	11	"""Summarize an access-level list (push/merge/unprotect) into one string."""
	12	return ", ".join(e.get("access_level_description", "") for e in entries) or "(none)"
	13
	14
	15	def branch_protections(group, cfg, projects):
	16	rows = []
	17	for p in projects:
	18	try:
	19	protected = paginate(
	20	f"{cfg['base_url']}/projects/{p['id']}/protected_branches", cfg
	21	)
	22	except requests.HTTPError as e:
	23	if e.response is not None and e.response.status_code in (403, 404):
	24	print(
	25	f" Skipping {p.get('path_with_namespace', p['id'])}: "
	26	f"protected_branches returned {e.response.status_code}",
	27	file=sys.stderr,
	28	)
	29	continue
	30	raise
	31	for b in protected:
	32	rows.append(
	33	{
	34	"project": p.get("path_with_namespace", ""),
	35	"branch": b.get("name", ""),
	36	"push_access": _levels(b.get("push_access_levels", [])),
	37	"merge_access": _levels(b.get("merge_access_levels", [])),
	38	"allow_force_push": b.get("allow_force_push"),
	39	"code_owner_approval_required": b.get(
	40	"code_owner_approval_required"
	41	),
	42	}
	43	)
	44	return rows

applications/gitlab/collectors/members.py added +72

@@ -0,0 +1,72 @@
	1	"""
	2	Collect group and project membership with access levels.
	3
	4	GitLab access levels:
	5	0 No access 5 Minimal 10 Guest 15 Planner
	6	20 Reporter 30 Developer 40 Maintainer 50 Owner 60 Admin
	7	"""
	8
	9	import sys
	10
	11	import requests
	12
	13	from .api import enc, paginate
	14
	15	ACCESS_LEVELS = {
	16	0: "No access",
	17	5: "Minimal",
	18	10: "Guest",
	19	15: "Planner",
	20	20: "Reporter",
	21	30: "Developer",
	22	40: "Maintainer",
	23	50: "Owner",
	24	60: "Admin",
	25	}
	26
	27
	28	def _role(level):
	29	return ACCESS_LEVELS.get(level, str(level))
	30
	31
	32	def group_members(group, cfg):
	33	"""Group members, including those inherited from parent groups."""
	34	members = paginate(f"{cfg['base_url']}/groups/{enc(group)}/members/all", cfg)
	35	return [
	36	{
	37	"username": m["username"],
	38	"name": m.get("name", ""),
	39	"access_level": m["access_level"],
	40	"role": _role(m["access_level"]),
	41	"state": m.get("state", ""),
	42	}
	43	for m in members
	44	]
	45
	46
	47	def project_members(group, cfg, projects):
	48	"""Direct and inherited members of every project in the group."""
	49	rows = []
	50	for p in projects:
	51	try:
	52	members = paginate(f"{cfg['base_url']}/projects/{p['id']}/members/all", cfg)
	53	except requests.HTTPError as e:
	54	if e.response is not None and e.response.status_code in (403, 404):
	55	print(
	56	f" Skipping {p.get('path_with_namespace', p['id'])}: "
	57	f"members returned {e.response.status_code}",
	58	file=sys.stderr,
	59	)
	60	continue
	61	raise
	62	for m in members:
	63	rows.append(
	64	{
	65	"project": p.get("path_with_namespace", ""),
	66	"username": m["username"],
	67	"name": m.get("name", ""),
	68	"access_level": m["access_level"],
	69	"role": _role(m["access_level"]),
	70	}
	71	)
	72	return rows

applications/gitlab/collectors/pipelines.py added +38

@@ -0,0 +1,38 @@
	1	"""Collect CI/CD pipeline history for every project in the group."""
	2
	3	import sys
	4
	5	import requests
	6
	7	from .api import paginate
	8
	9
	10	def pipelines(group, cfg, projects):
	11	rows = []
	12	for p in projects:
	13	try:
	14	project_pipelines = paginate(
	15	f"{cfg['base_url']}/projects/{p['id']}/pipelines", cfg
	16	)
	17	except requests.HTTPError as e:
	18	if e.response is not None and e.response.status_code in (403, 404):
	19	print(
	20	f" Skipping {p.get('path_with_namespace', p['id'])}: "
	21	f"pipelines returned {e.response.status_code}",
	22	file=sys.stderr,
	23	)
	24	continue
	25	raise
	26	for pipe in project_pipelines:
	27	rows.append(
	28	{
	29	"project": p.get("path_with_namespace", ""),
	30	"pipeline_id": pipe.get("id"),
	31	"status": pipe.get("status", ""),
	32	"ref": pipe.get("ref", ""),
	33	"source": pipe.get("source", ""),
	34	"created_at": pipe.get("created_at", ""),
	35	"web_url": pipe.get("web_url", ""),
	36	}
	37	)
	38	return rows

applications/gitlab/collectors/projects.py added +33

@@ -0,0 +1,33 @@
	1	"""
	2	Enumerate the projects in a GitLab group.
	3
	4	fetch_projects() returns the raw project objects once; the per-project
	5	collectors reuse that cache to avoid re-listing the group.
	6	"""
	7
	8	from .api import enc, paginate
	9
	10
	11	def fetch_projects(group, cfg):
	12	"""List all projects in the group, including subgroups."""
	13	return paginate(
	14	f"{cfg['base_url']}/groups/{enc(group)}/projects",
	15	cfg,
	16	{"include_subgroups": "true", "archived": "false"},
	17	)
	18
	19
	20	def project_list(group, cfg, projects):
	21	"""Format the project cache into audit rows."""
	22	return [
	23	{
	24	"id": p["id"],
	25	"name": p["name"],
	26	"path": p.get("path_with_namespace", ""),
	27	"visibility": p.get("visibility", ""),
	28	"default_branch": p.get("default_branch", ""),
	29	"archived": p.get("archived", False),
	30	"web_url": p.get("web_url", ""),
	31	}
	32	for p in projects
	33	]

applications/gitlab/collectors/settings.py added +38

@@ -0,0 +1,38 @@
	1	"""
	2	Collect the instance password policy from application settings.
	3
	4	Requires an admin token on a self-hosted instance; not available on
	5	GitLab.com. Returns an empty list with a warning on 403/404.
	6	"""
	7
	8	import sys
	9
	10	import requests
	11
	12	PASSWORD_FIELDS = [
	13	"minimum_password_length",
	14	"password_number_required",
	15	"password_symbol_required",
	16	"password_uppercase_required",
	17	"password_lowercase_required",
	18	]
	19
	20
	21	def password_policy(group, cfg):
	22	"""group is unused; application settings are instance-wide."""
	23	url = f"{cfg['base_url']}/application/settings"
	24	try:
	25	resp = requests.get(url, headers=cfg["headers"], timeout=cfg["timeout"])
	26	resp.raise_for_status()
	27	except requests.HTTPError as e:
	28	if e.response is not None and e.response.status_code in (403, 404):
	29	print(
	30	"Warning: application settings require an admin token on a "
	31	"self-hosted instance -- skipping.",
	32	file=sys.stderr,
	33	)
	34	return []
	35	raise
	36
	37	settings = resp.json()
	38	return [{field: settings.get(field, "Not set") for field in PASSWORD_FIELDS}]

applications/gitlab/config.py added +45

@@ -0,0 +1,45 @@
	1	"""
	2	Configuration loader for the GitLab audit tool.
	3
	4	Reads GITLAB_TOKEN, GITLAB_GROUP, and (optionally) GITLAB_URL from the
	5	environment.
	6
	7	Usage:
	8	export GITLAB_TOKEN=your_token
	9	export GITLAB_GROUP=your_group_id_or_path
	10	export GITLAB_URL=https://gitlab.example.com/api/v4 # self-hosted only
	11	"""
	12
	13	import os
	14	import sys
	15
	16	from collectors.api import DEFAULT_BASE_URL
	17
	18
	19	def load(group_override=None, base_url_override=None):
	20	"""Return a config dict. Exits with an error if required values are missing."""
	21	token = os.environ.get("GITLAB_TOKEN", "").strip()
	22	group = group_override or os.environ.get("GITLAB_GROUP", "").strip()
	23	base_url = (
	24	base_url_override
	25	or os.environ.get("GITLAB_URL", "").strip()
	26	or DEFAULT_BASE_URL
	27	)
	28
	29	missing = []
	30	if not token:
	31	missing.append("GITLAB_TOKEN")
	32	if not group:
	33	missing.append("GITLAB_GROUP (or pass --group)")
	34
	35	if missing:
	36	print(f"Error: missing required values: {', '.join(missing)}", file=sys.stderr)
	37	sys.exit(1)
	38
	39	return {
	40	"token": token,
	41	"group": group,
	42	"base_url": base_url.rstrip("/"),
	43	"headers": {"PRIVATE-TOKEN": token},
	44	"timeout": 30,
	45	}

applications/gitlab/passwords.py deleted −39

	@@ -1,39 +0,0 @@
1	"""
2	Verify if password policies are enforced in a self-hosted GitLab instance.
3
4	Ref: https://docs.gitlab.com/api/settings/
5	"""
6
7	import requests
8
9	BASE_URL = "https://gitlab.com/api/v4"
10	PRIVATE_TOKEN = "your_access_token"
11	TIMEOUT = 30
12
13	URL = f"{BASE_URL}/application/settings"
14	HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN}
15
16	if __name__ == "__main__":
17	# Get application settings
18	response = requests.get(URL, headers=HEADERS, timeout=TIMEOUT)
19	if response.status_code == 200:
20	settings = response.json()
21	minimum_password_length = settings.get("minimum_password_length", "Not set")
22	password_number_required = settings.get("password_number_required", "Not set")
23	password_symbol_required = settings.get("password_symbol_required", "Not set")
24	password_uppercase_required = settings.get(
25	"password_uppercase_required", "Not set"
26	)
27	password_lowercase_required = settings.get(
28	"password_lowercase_required", "Not set"
29	)
30
31	print(f"Password Length: {minimum_password_length}")
32	print(f"Password Number Required: {password_number_required}")
33	print(f"Password Symbol Required: {password_symbol_required}")
34	print(f"Password Uppercase Required: {password_uppercase_required}")
35	print(f"Password Lowercase Required: {password_lowercase_required}")
36	else:
37	print(
38	f"Failed to fetch application settings: {response.status_code}, {response.text}"
39	)

applications/gitlab/pipelines.py deleted −59

	@@ -1,59 +0,0 @@
1	"""
2	Review CI/CD pipelines and their configurations for a specific GitLab project.
3	"""
4
5	import requests
6
7	BASE_URL = "https://gitlab.com/api/v4"
8	PRIVATE_TOKEN = "your_access_token"
9	PROJECT_ID = "project_id"
10	TIMEOUT = 30
11
12	HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN}
13
14	if __name__ == "__main__":
15	page = 1
16	per_page = 100
17
18	while True:
19	response = requests.get(
20	f"{BASE_URL}/projects/{PROJECT_ID}/pipelines",
21	headers=HEADERS,
22	params={"page": page, "per_page": per_page},
23	timeout=TIMEOUT,
24	)
25	if response.status_code == 200:
26	pipelines = response.json()
27	if not pipelines:
28	break
29
30	for pipeline in pipelines:
31	pipeline_id = pipeline["id"]
32	status = pipeline["status"]
33	ref = pipeline["ref"]
34	created_at = pipeline["created_at"]
35	duration = pipeline.get("duration", "N/A")
36
37	print(f"Pipeline ID: {pipeline_id}")
38	print(f" Status: {status}")
39	print(f" Ref: {ref}")
40	print(f" Created At: {created_at}")
41	print(f" Duration: {duration} seconds")
42
43	detail_response = requests.get(
44	f"{BASE_URL}/projects/{PROJECT_ID}/pipelines/{pipeline_id}",
45	headers=HEADERS,
46	timeout=TIMEOUT,
47	)
48	if detail_response.status_code == 200:
49	pipeline_details = detail_response.json()
50	print(f" Configuration: {pipeline_details.get('config', 'N/A')}")
51	else:
52	print(
53	f" Failed to fetch pipeline details: {detail_response.status_code}, {detail_response.text}"
54	)
55
56	page += 1
57	else:
58	print(f"Failed to fetch pipelines: {response.status_code}, {response.text}")
59	break

applications/gitlab/provisioning.py deleted −32

	@@ -1,32 +0,0 @@
1	"""
2	Track user creation and deletion events in GitLab with timestamps.
3	"""
4
5	import requests
6
7	BASE_URL = "https://gitlab.com/api/v4"
8	PRIVATE_TOKEN = "your_access_token"
9	GROUP_ID = "your_group_id"
10	TIMEOUT = 30
11
12	URL = f"{BASE_URL}/groups/{GROUP_ID}/audit_events"
13	HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN}
14
15	if __name__ == "__main__":
16	# Get audit events
17	response = requests.get(URL, headers=HEADERS, timeout=TIMEOUT)
18	if response.status_code == 200:
19	audit_events = response.json()
20	for event in audit_events:
21	if event["entity_type"] == "User" or event["entity_type"] == "Group":
22	action = event["event_name"]
23	member_id = event["details"].get("member_id")
24	created_at = event["created_at"]
25	author = event["author_id"]
26	if action in ["member_created", "member_destroyed", "member_updated"]:
27	print(
28	f"Group: {GROUP_ID}\n",
29	f" {created_at} : Action: {action}, Member: {member_id}, Author: {author}",
30	)
31	else:
32	print(f"Failed to fetch audit events: {response.status_code}, {response.text}")

applications/gitlab/reporters/__init__.py added

applications/gitlab/reporters/csv_reporter.py added +46

@@ -0,0 +1,46 @@
	1	"""CSV reporter: writes one CSV file per data section into an output directory."""
	2
	3	import csv
	4	import os
	5
	6
	7	def write(output_dir, filename, rows):
	8	"""
	9	Write a list of dicts to a CSV file in output_dir.
	10	Skips writing if rows is empty, but logs the skip.
	11	"""
	12	if not rows:
	13	print(f" {filename}: no data, skipping")
	14	return
	15
	16	os.makedirs(output_dir, exist_ok=True)
	17	path = os.path.join(output_dir, filename)
	18
	19	with open(path, "w", newline="", encoding="utf-8") as f:
	20	writer = csv.DictWriter(f, fieldnames=rows[0].keys())
	21	writer.writeheader()
	22	writer.writerows(rows)
	23
	24	print(f" {filename}: {len(rows)} rows -> {path}")
	25
	26
	27	def write_summary(output_dir, group, sections):
	28	"""
	29	Write a plain-text summary file listing section names and row counts.
	30	sections: list of (label, row_count) tuples
	31	"""
	32	path = os.path.join(output_dir, "summary.txt")
	33	lines = [
	34	"GitLab Audit Package",
	35	f"Group: {group}",
	36	"",
	37	"Section Rows",
	38	f"{'─' * 40}",
	39	]
	40	for label, count in sections:
	41	lines.append(f"{label:<35}{count}")
	42
	43	with open(path, "w", encoding="utf-8") as f:
	44	f.write("\n".join(lines) + "\n")
	45
	46	print(f" summary.txt -> {path}")

applications/gitlab/repositories.py deleted −51

	@@ -1,51 +0,0 @@
1	"""
2	List all repositories (projects) for a user or organization in GitLab.
3	"""
4
5	import requests
6
7	BASE_URL = "https://gitlab.com/api/v4"
8	PRIVATE_TOKEN = "your_access_token"
9	USER_ID = "your_user_or_group_id"
10	TIMEOUT = 30
11
12	URL = f"{BASE_URL}/groups/{USER_ID}/projects" # Group URL
13	# URL = f"{BASE_URL}/users/{USER_ID}/projects" # User URL
14	HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN}
15
16
17	def list_projects(user_or_group_id):
18	PER_PAGE = 100
19	page = 1
20	projects = []
21
22	while True:
23	response = requests.get(
24	URL,
25	headers=HEADERS,
26	timeout=TIMEOUT,
27	params={"page": page, "per_page": PER_PAGE},
28	)
29
30	if response.status_code == 200:
31	current_projects = response.json()
32	if not current_projects:
33	break
34	projects.extend(current_projects)
35	page += 1
36	else:
37	print(
38	f"Failed to retrieve projects: {response.status_code} - {response.text}"
39	)
40	break
41
42	if projects:
43	print(f"Projects under ID: {user_or_group_id}:")
44	for project in projects:
45	print(f"- {project['name']} (ID: {project['id']})")
46	else:
47	print(f"No projects found for ID: {user_or_group_id}.")
48
49
50	if __name__ == "__main__":
51	list_projects(USER_ID)

applications/gitlab/users.py deleted −53

	@@ -1,53 +0,0 @@
1	"""
2	Gather all members of specified GitLab groups and projects and their access levels.
3
4	Ref: https://docs.gitlab.com/api/members/
5	"""
6
7	import requests
8
9	BASE_URL = "https://gitlab.com/api/v4"
10	PRIVATE_TOKEN = "your_access_token"
11	GROUP_IDS = ["group_id_1", "group_id_2"] # Add your group IDs here
12	PROJECT_IDS = ["project_id_1", "project_id_2"] # Add your project IDs here
13	TIMEOUT = 30
14
15	HEADERS = {"PRIVATE-TOKEN": PRIVATE_TOKEN}
16
17
18	def get_members(url, name):
19	response = requests.get(url, headers=HEADERS, timeout=TIMEOUT)
20	if response.status_code == 200:
21	members = response.json()
22	print(f"\n{name} Members:")
23	for member in members:
24	print(
25	f"Username: {member['username']}, Access Level: {member['access_level']}"
26	)
27	else:
28	print(
29	f"Failed to fetch members for {name}: {response.status_code}, {response.text}"
30	)
31
32
33	if __name__ == "__main__":
34	access_levels = """Access Level Roles:
35	0 : No access
36	5 : Minimal access
37	10 : Guest
38	15 : Planner
39	20 : Reporter
40	30 : Developer
41	40 : Maintainer
42	50 : Owner
43	60 : Admin
44	"""
45	print(access_levels)
46
47	for group_id in GROUP_IDS:
48	group_url = f"{BASE_URL}/groups/{group_id}/members"
49	get_members(group_url, f"Group {group_id}")
50
51	for project_id in PROJECT_IDS:
52	project_url = f"{BASE_URL}/projects/{project_id}/members"
53	get_members(project_url, f"Project {project_id}")

audit_tui.py added +7

@@ -0,0 +1,7 @@
	1	#!/usr/bin/env python3
	2	"""Launch the Audit Tools interactive terminal UI."""
	3
	4	from tui.app import main
	5
	6	if __name__ == "__main__":
	7	main()

conftest.py added +10

@@ -0,0 +1,10 @@
	1	"""Ensure the repository root is importable so tests can use absolute imports
	2	(``from tui import ...``, ``from sampling.sampling_tool import ...``) regardless
	3	of how pytest is invoked."""
	4
	5	import os
	6	import sys
	7
	8	ROOT = os.path.dirname(os.path.abspath(__file__))
	9	if ROOT not in sys.path:
	10	sys.path.insert(0, ROOT)

requirements.txt +1

	@@ -4,6 +4,7 @@ xlrd
4	PyYAML	4	PyYAML
5	pytest	5	pytest
6	requests	6	requests
		7	textual
7	dash	8	dash
8	plotly	9	plotly
9	urllib3>=2.7.0	10	urllib3>=2.7.0

tui/README.md added +63

@@ -0,0 +1,63 @@
	1	# Audit Tools — Interactive TUI
	2
	3	A terminal UI that walks you through running an audit. It presents a platform
	4	menu, collects connection details and check selection, then runs the existing
	5	collectors with live progress.
	6
	7	GitHub and GitLab are supported. Adding a platform is a matter of writing a
	8	runner and a `Platform` descriptor in `tui/platforms.py` — the screens are
	9	platform-agnostic.
	10
	11	## Run it
	12
	13	```bash
	14	pip install -r requirements.txt
	15	python audit_tui.py
	16	```
	17
	18	The connection fields are pre-filled from environment variables if set:
	19
	20	```bash
	21	# GitHub
	22	export GITHUB_ORG=my-org
	23	export GITHUB_TOKEN=ghp_... # needs read:org and repo scopes
	24
	25	# GitLab
	26	export GITLAB_GROUP=my-group
	27	export GITLAB_TOKEN=glpat-... # needs read_api scope
	28	export GITLAB_URL=https://gitlab.example.com/api/v4 # self-hosted only
	29	```
	30
	31	## Walkthrough
	32
	33	1. Platform — choose GitHub or GitLab.
	34	2. Connection — the audit subject (org / group), a masked token, and any
	35	platform-specific fields (branch for GitHub; API base URL for GitLab).
	36	3. Checks — toggle which checks to run. Plan-restricted checks (GitHub's
	37	Enterprise audit log; GitLab's Premium and self-hosted checks) are off by
	38	default.
	39	4. Run — a progress bar and live log show each check completing with its row
	40	count. Errors on a single check are reported without stopping the run.
	41
	42	## Output
	43
	44	The TUI writes the same package the platform's `audit.py` produces:
	45	`<output>/github_audit_<org>_<date>/` or `<output>/gitlab_audit_<group>_<date>/`,
	46	one CSV per check plus a `summary.txt`. It reuses each platform's collectors and
	47	CSV reporter unchanged — the TUI is only an interactive driver around them.
	48
	49	## Keys
	50
	51	- `Esc` — back / return to menu
	52	- `Ctrl+P` — command palette
	53	- `q` — quit (from the menu)
	54
	55	## Tests
	56
	57	```bash
	58	python -m pytest tui/tests
	59	```
	60
	61	The tests stub the network-bound collectors, so they run offline: one suite
	62	covers the run orchestration, another drives the app headlessly through every
	63	screen.

tui/__init__.py added +1

	@@ -0,0 +1 @@
		1	"""Interactive terminal UI for running audit collectors."""

tui/app.py added +337

@@ -0,0 +1,337 @@
	1	"""
	2	Audit Tools — interactive terminal UI.
	3
	4	Presents a platform menu, walks the user through credentials and check
	5	selection, then runs the selected platform's collectors with live progress.
	6
	7	Run it with:
	8
	9	python audit_tui.py
	10	"""
	11
	12	from typing import ClassVar
	13
	14	from rich.text import Text
	15	from textual import work
	16	from textual.app import App, ComposeResult
	17	from textual.containers import Center, Horizontal, Vertical
	18	from textual.screen import Screen
	19	from textual.widgets import (
	20	Button,
	21	Footer,
	22	Header,
	23	Input,
	24	Label,
	25	ProgressBar,
	26	RichLog,
	27	SelectionList,
	28	Static,
	29	)
	30	from textual.widgets.selection_list import Selection
	31
	32	from tui import platforms
	33	from tui.common import Check, ProgressEvent
	34
	35
	36	class MenuScreen(Screen):
	37	"""Pick a platform to audit."""
	38
	39	BINDINGS: ClassVar[list] = [("q", "app.quit", "Quit")]
	40
	41	def compose(self) -> ComposeResult:
	42	yield Header()
	43	with Center(), Vertical(id="menu-box"):
	44	yield Static("Select a platform to audit", classes="prompt")
	45	for platform in platforms.PLATFORMS:
	46	label = platform.label
	47	if not platform.enabled:
	48	label = f"{label} — coming soon"
	49	yield Button(
	50	label,
	51	id=platform.key,
	52	variant="primary" if platform.enabled else "default",
	53	disabled=not platform.enabled,
	54	)
	55	yield Footer()
	56
	57	def on_mount(self) -> None:
	58	self.sub_title = "Select a platform"
	59
	60	def on_button_pressed(self, event: Button.Pressed) -> None:
	61	for platform in platforms.PLATFORMS:
	62	if event.button.id == platform.key and platform.enabled:
	63	self.app.platform = platform
	64	self.app.push_screen(ConfigScreen())
	65	return
	66
	67
	68	class ConfigScreen(Screen):
	69	"""Collect the connection details for the chosen platform."""
	70
	71	BINDINGS: ClassVar[list] = [("escape", "back", "Back")]
	72
	73	def compose(self) -> ComposeResult:
	74	platform = self.app.platform
	75	yield Header()
	76	with Center(), Vertical(id="form-box"):
	77	yield Static(
	78	f"{platform.label} audit — connection details", classes="prompt"
	79	)
	80	for f in platform.fields:
	81	yield Label(f.label)
	82	yield Input(
	83	value=platforms.prefill(f),
	84	placeholder=f.placeholder,
	85	password=f.password,
	86	id=f.key,
	87	)
	88	yield Static("", id="form-error", classes="error")
	89	with Horizontal(classes="buttons"):
	90	yield Button("Back", id="back")
	91	yield Button("Continue", id="continue", variant="primary")
	92	yield Footer()
	93
	94	def on_mount(self) -> None:
	95	platform = self.app.platform
	96	self.sub_title = f"{platform.label} · connection"
	97	self.query_one(f"#{platform.fields[0].key}", Input).focus()
	98
	99	def action_back(self) -> None:
	100	self.app.pop_screen()
	101
	102	def on_button_pressed(self, event: Button.Pressed) -> None:
	103	if event.button.id == "back":
	104	self.app.pop_screen()
	105	elif event.button.id == "continue":
	106	self._submit()
	107
	108	def on_input_submitted(self, event: Input.Submitted) -> None:
	109	self._submit()
	110
	111	def _submit(self) -> None:
	112	platform = self.app.platform
	113	settings = {}
	114	missing = []
	115	for f in platform.fields:
	116	value = self.query_one(f"#{f.key}", Input).value.strip()
	117	if not value:
	118	value = f.default
	119	if f.required and not value:
	120	missing.append(f.label.lower())
	121	settings[f.key] = value
	122
	123	if missing:
	124	self.query_one("#form-error", Static).update(
	125	f"Please provide: {', '.join(missing)}."
	126	)
	127	return
	128
	129	self.app.settings = settings
	130	self.app.push_screen(ChecksScreen())
	131
	132
	133	class ChecksScreen(Screen):
	134	"""Choose which checks to run."""
	135
	136	BINDINGS: ClassVar[list] = [("escape", "back", "Back")]
	137
	138	def compose(self) -> ComposeResult:
	139	platform = self.app.platform
	140	yield Header()
	141	with Center(), Vertical(id="checks-box"):
	142	yield Static("Select checks to run", classes="prompt")
	143	yield SelectionList(
	144	*[
	145	Selection(
	146	self._prompt(c),
	147	c.key,
	148	c.key in platform.default_selection,
	149	)
	150	for c in platform.checks
	151	],
	152	id="checks",
	153	)
	154	yield Static("", id="checks-error", classes="error")
	155	with Horizontal(classes="buttons"):
	156	yield Button("Back", id="back")
	157	yield Button("Run audit", id="run", variant="primary")
	158	yield Footer()
	159
	160	def on_mount(self) -> None:
	161	self.sub_title = f"{self.app.platform.label} · select checks"
	162	self.query_one("#checks", SelectionList).focus()
	163
	164	@staticmethod
	165	def _prompt(check: Check) -> Text:
	166	text = Text(check.label)
	167	if check.note:
	168	text.append(f" ({check.note})", style="dim italic")
	169	return text
	170
	171	def action_back(self) -> None:
	172	self.app.pop_screen()
	173
	174	def on_button_pressed(self, event: Button.Pressed) -> None:
	175	if event.button.id == "back":
	176	self.app.pop_screen()
	177	elif event.button.id == "run":
	178	selected = list(self.query_one("#checks", SelectionList).selected)
	179	if not selected:
	180	self.query_one("#checks-error", Static).update(
	181	"Select at least one check."
	182	)
	183	return
	184	self.app.selected_keys = selected
	185	self.app.push_screen(RunScreen())
	186
	187
	188	class RunScreen(Screen):
	189	"""Run the selected checks with live progress."""
	190
	191	BINDINGS: ClassVar[list] = [("escape", "home", "Menu")]
	192
	193	def compose(self) -> ComposeResult:
	194	yield Header()
	195	with Vertical(id="run-box"):
	196	yield Static(id="run-target", classes="prompt")
	197	yield ProgressBar(id="progress", show_eta=False)
	198	yield RichLog(id="log", markup=True, highlight=False, wrap=True)
	199	with Horizontal(classes="buttons"):
	200	yield Button("Back to menu", id="menu", disabled=True)
	201	yield Button("Quit", id="quit", disabled=True, variant="primary")
	202	yield Footer()
	203
	204	def on_mount(self) -> None:
	205	platform = self.app.platform
	206	settings = self.app.settings
	207	keys = self.app.selected_keys
	208	self.sub_title = f"{platform.label} · running"
	209	self.output_dir = platform.output_dir(settings)
	210	target = settings[platform.id_key]
	211	self.query_one("#run-target", Static).update(
	212	f"Auditing [b]{target}[/] · {len(keys)} checks · → {self.output_dir}"
	213	)
	214	self.query_one("#progress", ProgressBar).update(total=len(keys), progress=0)
	215	self.run_audit()
	216
	217	@work(thread=True)
	218	def run_audit(self) -> None:
	219	platform = self.app.platform
	220	settings = self.app.settings
	221	keys = self.app.selected_keys
	222	try:
	223	platform.run(
	224	settings,
	225	self.output_dir,
	226	keys,
	227	lambda ev: self.app.call_from_thread(self._handle_event, ev),
	228	)
	229	except Exception as e: # noqa: BLE001 - report unexpected failures in the UI
	230	self.app.call_from_thread(self._log, f"[red]Run failed:[/] {e}")
	231	finally:
	232	self.app.call_from_thread(self._finish)
	233
	234	def _log(self, markup: str) -> None:
	235	self.query_one("#log", RichLog).write(markup)
	236
	237	def _handle_event(self, ev: ProgressEvent) -> None:
	238	if ev.kind == "fetch":
	239	self._log(f"[dim]· {ev.label}…[/]")
	240	elif ev.kind == "start":
	241	self._log(f"[cyan]▶[/] {ev.label}…")
	242	elif ev.kind == "done":
	243	self._log(f"[green]✓[/] {ev.label} — [b]{ev.count}[/] rows")
	244	self.query_one("#progress", ProgressBar).advance(1)
	245	elif ev.kind == "error":
	246	self._log(f"[red]✗[/] {ev.label} — {ev.message}")
	247	self.query_one("#progress", ProgressBar).advance(1)
	248	elif ev.kind == "summary":
	249	self._log("")
	250	self._log(f"[bold green]Done.[/] Package written to {ev.label}")
	251
	252	def _finish(self) -> None:
	253	self.query_one("#menu", Button).disabled = False
	254	self.query_one("#quit", Button).disabled = False
	255
	256	def action_home(self) -> None:
	257	self.app.show_menu()
	258
	259	def on_button_pressed(self, event: Button.Pressed) -> None:
	260	if event.button.id == "menu":
	261	self.app.show_menu()
	262	elif event.button.id == "quit":
	263	self.app.exit()
	264
	265
	266	class AuditApp(App):
	267	TITLE = "Audit Tools"
	268
	269	CSS = """
	270	Screen {
	271	align: center middle;
	272	}
	273	#menu-box, #form-box, #checks-box {
	274	width: 64;
	275	height: auto;
	276	padding: 1 2;
	277	border: round $primary;
	278	}
	279	#run-box {
	280	width: 90%;
	281	height: 90%;
	282	padding: 1 2;
	283	border: round $primary;
	284	}
	285	.prompt {
	286	text-style: bold;
	287	margin-bottom: 1;
	288	}
	289	.error {
	290	color: $error;
	291	margin-top: 1;
	292	}
	293	Label {
	294	margin-top: 1;
	295	}
	296	.buttons {
	297	height: auto;
	298	margin-top: 1;
	299	align-horizontal: right;
	300	}
	301	.buttons Button {
	302	margin-left: 2;
	303	}
	304	#menu-box Button {
	305	width: 100%;
	306	margin-top: 1;
	307	}
	308	#checks {
	309	height: auto;
	310	max-height: 14;
	311	}
	312	#log {
	313	height: 1fr;
	314	border: round $panel;
	315	padding: 0 1;
	316	margin-top: 1;
	317	}
	318	"""
	319
	320	def on_mount(self) -> None:
	321	self.platform = None
	322	self.settings: dict = {}
	323	self.selected_keys: list = []
	324	self.push_screen(MenuScreen())
	325
	326	def show_menu(self) -> None:
	327	"""Pop back to the platform menu."""
	328	while len(self.screen_stack) > 2:
	329	self.pop_screen()
	330
	331
	332	def main() -> None:
	333	AuditApp().run()
	334
	335
	336	if __name__ == "__main__":
	337	main()

tui/common.py added +30

@@ -0,0 +1,30 @@
	1	"""Shared types used by the platform runners and the TUI."""
	2
	3	from collections.abc import Callable
	4	from dataclasses import dataclass
	5
	6
	7	# ``arg`` describes how a collector is called:
	8	# "base" -> fn(target, cfg)
	9	# "collabs" -> fn(target, cfg, repo_collabs) (GitHub collaborator cache)
	10	# "projects" -> fn(target, cfg, projects) (GitLab project cache)
	11	@dataclass(frozen=True)
	12	class Check:
	13	key: str
	14	label: str
	15	fn: Callable
	16	filename: str
	17	arg: str = "base"
	18	note: str = ""
	19
	20
	21	# kind is one of: "fetch", "start", "done", "error", "summary".
	22	@dataclass(frozen=True)
	23	class ProgressEvent:
	24	kind: str
	25	label: str
	26	count: int \| None = None
	27	message: str = ""
	28
	29
	30	ProgressCallback = Callable[[ProgressEvent], None]

tui/github_runner.py added +180

@@ -0,0 +1,180 @@
	1	"""
	2	Drive the existing GitHub audit collectors from the TUI.
	3
	4	This module reuses the collectors and CSV reporter under
	5	``applications/github`` unchanged. It exposes:
	6
	7	- ``CHECKS``: the list of available audit checks the UI presents.
	8	- ``run_audit``: run the selected checks, write the same output package
	9	``audit.py`` produces, and report progress through a callback.
	10	"""
	11
	12	import os
	13	import sys
	14	from collections.abc import Iterable
	15	from datetime import date
	16
	17	from tui.common import Check, ProgressCallback, ProgressEvent
	18
	19	# Import the GitHub collectors as a namespaced package so the GitHub and GitLab
	20	# collector packages (both named ``collectors`` on disk) can coexist in one
	21	# process. Requires the repo root on sys.path.
	22	_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
	23	if _REPO_ROOT not in sys.path:
	24	sys.path.insert(0, _REPO_ROOT)
	25
	26	from applications.github.collectors import (
	27	audit_log,
	28	branch_protections,
	29	commits,
	30	members,
	31	)
	32	from applications.github.reporters import csv_reporter
	33
	34	# --- Check registry ---------------------------------------------------------
	35
	36	# For GitHub, ``arg`` is "base" -> fn(org, cfg), "collabs" -> fn(org, cfg,
	37	# repo_collabs), or "branch" -> fn(org, cfg, branch).
	38
	39	CHECKS: list[Check] = [
	40	Check("member_roster", "Member roster", members.member_roster, "member_roster.csv"),
	41	Check(
	42	"two_factor",
	43	"2FA disabled",
	44	members.two_factor_disabled,
	45	"two_factor_disabled.csv",
	46	note="requires org owner token",
	47	),
	48	Check(
	49	"outside_collaborators",
	50	"Outside collaborators",
	51	members.outside_collaborators,
	52	"outside_collaborators.csv",
	53	arg="collabs",
	54	),
	55	Check(
	56	"privileged_access",
	57	"Privileged access",
	58	members.privileged_access,
	59	"privileged_access.csv",
	60	arg="collabs",
	61	),
	62	Check(
	63	"pending_invitations",
	64	"Pending invitations",
	65	members.pending_invitations,
	66	"pending_invitations.csv",
	67	),
	68	Check(
	69	"team_permissions",
	70	"Team permissions",
	71	members.team_permissions,
	72	"team_permissions.csv",
	73	),
	74	Check(
	75	"permission_matrix",
	76	"Permission matrix",
	77	members.permission_matrix,
	78	"permission_matrix.csv",
	79	arg="collabs",
	80	),
	81	Check(
	82	"branch_protections",
	83	"Branch protections",
	84	branch_protections.branch_protections,
	85	"branch_protections.csv",
	86	),
	87	Check("commits", "Commits", commits.commits, "commits.csv", arg="branch"),
	88	Check(
	89	"audit_log",
	90	"Audit log (branch/ruleset changes)",
	91	audit_log.audit_log,
	92	"audit_log.csv",
	93	note="requires GitHub Enterprise Cloud",
	94	),
	95	]
	96
	97	DEFAULT_SELECTION = [c.key for c in CHECKS if c.key != "audit_log"]
	98
	99
	100	# --- Config + output helpers ------------------------------------------------
	101
	102
	103	def build_cfg(token: str) -> dict:
	104	"""Build the config dict the collectors expect (mirrors config.load())."""
	105	return {
	106	"token": token,
	107	"headers": {
	108	"Authorization": f"token {token}",
	109	"Accept": "application/vnd.github.v3+json",
	110	},
	111	"timeout": 30,
	112	}
	113
	114
	115	def default_output_dir(out: str, org: str) -> str:
	116	"""Match the folder naming used by audit.py."""
	117	return os.path.join(out, f"github_audit_{org}_{date.today().isoformat()}")
	118
	119
	120	# --- Runner -----------------------------------------------------------------
	121
	122
	123	def run_audit(
	124	*,
	125	org: str,
	126	token: str,
	127	output_dir: str,
	128	branch: str,
	129	selected_keys: Iterable[str],
	130	on_event: ProgressCallback,
	131	) -> list[tuple[str, int]]:
	132	"""
	133	Run the selected checks and write the audit package to ``output_dir``.
	134
	135	A collector that raises is reported as an error and recorded with a count
	136	of 0, matching audit.py's behavior of never aborting the whole run.
	137
	138	Returns the list of (label, row_count) sections that was written to the
	139	summary file.
	140	"""
	141	cfg = build_cfg(token)
	142	selected = set(selected_keys)
	143	checks = [c for c in CHECKS if c.key in selected]
	144
	145	repo_collabs: list \| None = None
	146	if any(c.arg == "collabs" for c in checks):
	147	on_event(ProgressEvent("fetch", "Repo collaborators (shared cache)"))
	148	try:
	149	repo_collabs = members.fetch_repo_collaborators(org, cfg)
	150	except Exception as e: # noqa: BLE001 - surface, keep going
	151	on_event(
	152	ProgressEvent(
	153	"error", "Repo collaborators (shared cache)", message=str(e)
	154	)
	155	)
	156	repo_collabs = []
	157
	158	sections: list[tuple[str, int]] = []
	159	for c in checks:
	160	on_event(ProgressEvent("start", c.label))
	161	try:
	162	if c.arg == "collabs":
	163	rows = c.fn(org, cfg, repo_collabs or [])
	164	elif c.arg == "branch":
	165	rows = c.fn(org, cfg, branch)
	166	else:
	167	rows = c.fn(org, cfg)
	168	except Exception as e: # noqa: BLE001 - one bad check shouldn't kill the run
	169	on_event(ProgressEvent("error", c.label, message=str(e)))
	170	sections.append((c.label, 0))
	171	continue
	172
	173	csv_reporter.write(output_dir, c.filename, rows)
	174	sections.append((c.label, len(rows)))
	175	on_event(ProgressEvent("done", c.label, count=len(rows)))
	176
	177	csv_reporter.write_summary(output_dir, org, sections)
	178	total = sum(n for _, n in sections)
	179	on_event(ProgressEvent("summary", output_dir, count=total))
	180	return sections

tui/gitlab_runner.py added +163

@@ -0,0 +1,163 @@
	1	"""
	2	Drive the GitLab audit collectors from the TUI.
	3
	4	Reuses the collectors and CSV reporter under ``applications/gitlab`` unchanged.
	5	Mirrors github_runner: a ``CHECKS`` registry plus ``run_audit`` that writes the
	6	same package ``applications/gitlab/audit.py`` produces and reports progress
	7	through a callback.
	8	"""
	9
	10	import os
	11	import sys
	12	from collections.abc import Iterable
	13	from datetime import date
	14
	15	from tui.common import Check, ProgressCallback, ProgressEvent
	16
	17	# Namespaced import so the GitHub and GitLab collector packages (both named
	18	# ``collectors`` on disk) can coexist in one process.
	19	_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
	20	if _REPO_ROOT not in sys.path:
	21	sys.path.insert(0, _REPO_ROOT)
	22
	23	from applications.gitlab.collectors import (
	24	approvals,
	25	audit_events,
	26	branch_protections,
	27	members,
	28	pipelines,
	29	projects,
	30	settings,
	31	)
	32	from applications.gitlab.reporters import csv_reporter
	33
	34	# --- Check registry ---------------------------------------------------------
	35
	36	# For GitLab, ``arg`` is "base" -> fn(group, cfg) or "projects" -> fn(group,
	37	# cfg, projects), where the project cache is fetched once and shared.
	38
	39	CHECKS: list[Check] = [
	40	Check("group_members", "Group members", members.group_members, "group_members.csv"),
	41	Check(
	42	"projects", "Projects", projects.project_list, "projects.csv", arg="projects"
	43	),
	44	Check(
	45	"project_members",
	46	"Project members",
	47	members.project_members,
	48	"project_members.csv",
	49	arg="projects",
	50	),
	51	Check(
	52	"branch_protections",
	53	"Branch protections",
	54	branch_protections.branch_protections,
	55	"branch_protections.csv",
	56	arg="projects",
	57	),
	58	Check(
	59	"pipelines",
	60	"Pipelines",
	61	pipelines.pipelines,
	62	"pipelines.csv",
	63	arg="projects",
	64	),
	65	Check(
	66	"approval_rules",
	67	"Approval rules",
	68	approvals.approval_rules,
	69	"approval_rules.csv",
	70	arg="projects",
	71	note="requires Premium/Ultimate",
	72	),
	73	Check(
	74	"audit_events",
	75	"Audit events",
	76	audit_events.audit_events,
	77	"audit_events.csv",
	78	note="requires Premium/Ultimate",
	79	),
	80	Check(
	81	"password_policy",
	82	"Password policy",
	83	settings.password_policy,
	84	"password_policy.csv",
	85	note="self-hosted, admin token",
	86	),
	87	]
	88
	89	_PREMIUM = {"approval_rules", "audit_events", "password_policy"}
	90	DEFAULT_SELECTION = [c.key for c in CHECKS if c.key not in _PREMIUM]
	91
	92
	93	# --- Config + output helpers ------------------------------------------------
	94
	95
	96	def build_cfg(token: str, base_url: str) -> dict:
	97	"""Build the config dict the collectors expect (mirrors config.load())."""
	98	return {
	99	"token": token,
	100	"base_url": base_url.rstrip("/"),
	101	"headers": {"PRIVATE-TOKEN": token},
	102	"timeout": 30,
	103	}
	104
	105
	106	def default_output_dir(out: str, group: str) -> str:
	107	"""Match the folder naming used by applications/gitlab/audit.py."""
	108	safe_group = group.replace("/", "-")
	109	return os.path.join(out, f"gitlab_audit_{safe_group}_{date.today().isoformat()}")
	110
	111
	112	# --- Runner -----------------------------------------------------------------
	113
	114
	115	def run_audit(
	116	*,
	117	group: str,
	118	token: str,
	119	base_url: str,
	120	output_dir: str,
	121	selected_keys: Iterable[str],
	122	on_event: ProgressCallback,
	123	) -> list[tuple[str, int]]:
	124	"""
	125	Run the selected checks and write the audit package to ``output_dir``.
	126
	127	A collector that raises is reported as an error and recorded with a count
	128	of 0, so one bad check never aborts the whole run.
	129	"""
	130	cfg = build_cfg(token, base_url)
	131	selected = set(selected_keys)
	132	checks = [c for c in CHECKS if c.key in selected]
	133
	134	project_cache: list \| None = None
	135	if any(c.arg == "projects" for c in checks):
	136	on_event(ProgressEvent("fetch", "Projects (shared cache)"))
	137	try:
	138	project_cache = projects.fetch_projects(group, cfg)
	139	except Exception as e: # noqa: BLE001 - surface, keep going
	140	on_event(ProgressEvent("error", "Projects (shared cache)", message=str(e)))
	141	project_cache = []
	142
	143	sections: list[tuple[str, int]] = []
	144	for c in checks:
	145	on_event(ProgressEvent("start", c.label))
	146	try:
	147	if c.arg == "projects":
	148	rows = c.fn(group, cfg, project_cache or [])
	149	else:
	150	rows = c.fn(group, cfg)
	151	except Exception as e: # noqa: BLE001 - one bad check shouldn't kill the run
	152	on_event(ProgressEvent("error", c.label, message=str(e)))
	153	sections.append((c.label, 0))
	154	continue
	155
	156	csv_reporter.write(output_dir, c.filename, rows)
	157	sections.append((c.label, len(rows)))
	158	on_event(ProgressEvent("done", c.label, count=len(rows)))
	159
	160	csv_reporter.write_summary(output_dir, group, sections)
	161	total = sum(n for _, n in sections)
	162	on_event(ProgressEvent("summary", output_dir, count=total))
	163	return sections

tui/platforms.py added +144

@@ -0,0 +1,144 @@
	1	"""
	2	Platform descriptors that let the TUI drive any collector runner.
	3
	4	Each Platform declares its connection form (``fields``), its checks, and how to
	5	compute the output directory and run the audit. The screens in app.py are
	6	written against this interface, so adding a platform is data, not new UI.
	7	"""
	8
	9	import os
	10	from collections.abc import Callable
	11	from dataclasses import dataclass, field
	12
	13	from tui import github_runner, gitlab_runner
	14	from tui.common import Check
	15
	16
	17	@dataclass(frozen=True)
	18	class Field:
	19	"""One input on the connection screen."""
	20
	21	key: str
	22	label: str
	23	placeholder: str = ""
	24	default: str = ""
	25	password: bool = False
	26	required: bool = False
	27	env: str \| None = None # environment variable used to pre-fill the value
	28
	29
	30	@dataclass(frozen=True)
	31	class Platform:
	32	key: str
	33	label: str
	34	id_key: str # which field is the audit subject (org / group)
	35	fields: list[Field]
	36	checks: list[Check]
	37	default_selection: list[str]
	38	output_dir: Callable[[dict], str] # (settings) -> path
	39	run: Callable[..., object] # (settings, output_dir, selected_keys, on_event)
	40	enabled: bool = True
	41	note: str = field(default="")
	42
	43
	44	def _prefill(f: Field) -> str:
	45	if f.env:
	46	value = os.environ.get(f.env, "").strip()
	47	if value:
	48	return value
	49	return f.default
	50
	51
	52	def _github_output_dir(s: dict) -> str:
	53	return github_runner.default_output_dir(s["out"], s["org"])
	54
	55
	56	def _github_run(s: dict, output_dir, selected_keys, on_event):
	57	return github_runner.run_audit(
	58	org=s["org"],
	59	token=s["token"],
	60	output_dir=output_dir,
	61	branch=s["branch"],
	62	selected_keys=selected_keys,
	63	on_event=on_event,
	64	)
	65
	66
	67	def _gitlab_output_dir(s: dict) -> str:
	68	return gitlab_runner.default_output_dir(s["out"], s["group"])
	69
	70
	71	def _gitlab_run(s: dict, output_dir, selected_keys, on_event):
	72	return gitlab_runner.run_audit(
	73	group=s["group"],
	74	token=s["token"],
	75	base_url=s["base_url"],
	76	output_dir=output_dir,
	77	selected_keys=selected_keys,
	78	on_event=on_event,
	79	)
	80
	81
	82	GITHUB = Platform(
	83	key="github",
	84	label="GitHub",
	85	id_key="org",
	86	fields=[
	87	Field("org", "Organization", "my-org", required=True, env="GITHUB_ORG"),
	88	Field(
	89	"token",
	90	"Personal access token",
	91	"ghp_… (read:org, repo)",
	92	password=True,
	93	required=True,
	94	env="GITHUB_TOKEN",
	95	),
	96	Field("out", "Output directory", default="./output"),
	97	Field("branch", "Branch (for commit history)", default="main"),
	98	],
	99	checks=github_runner.CHECKS,
	100	default_selection=github_runner.DEFAULT_SELECTION,
	101	output_dir=_github_output_dir,
	102	run=_github_run,
	103	)
	104
	105	GITLAB = Platform(
	106	key="gitlab",
	107	label="GitLab",
	108	id_key="group",
	109	fields=[
	110	Field(
	111	"group",
	112	"Group ID or path",
	113	"e.g. 1234567 or my-group",
	114	required=True,
	115	env="GITLAB_GROUP",
	116	),
	117	Field(
	118	"token",
	119	"Personal access token",
	120	"glpat-… (read_api)",
	121	password=True,
	122	required=True,
	123	env="GITLAB_TOKEN",
	124	),
	125	Field(
	126	"base_url",
	127	"API base URL (self-hosted)",
	128	default="https://gitlab.com/api/v4",
	129	env="GITLAB_URL",
	130	),
	131	Field("out", "Output directory", default="./output"),
	132	],
	133	checks=gitlab_runner.CHECKS,
	134	default_selection=gitlab_runner.DEFAULT_SELECTION,
	135	output_dir=_gitlab_output_dir,
	136	run=_gitlab_run,
	137	)
	138
	139	PLATFORMS = [GITHUB, GITLAB]
	140
	141
	142	def prefill(f: Field) -> str:
	143	"""Public accessor for a field's pre-filled value (env var or default)."""
	144	return _prefill(f)

tui/tests/test_app.py added +127

@@ -0,0 +1,127 @@
	1	"""Headless smoke tests for the Textual app.
	2
	3	Drives the app through its screens with a Pilot, stubbing the network-bound
	4	run_audit so no real GitHub calls are made. Uses asyncio.run so the suite does
	5	not require the pytest-asyncio plugin.
	6	"""
	7
	8	import asyncio
	9
	10	from textual.widgets import Button, Input
	11
	12	from tui import github_runner as gh
	13	from tui.app import AuditApp, ChecksScreen, ConfigScreen, MenuScreen, RunScreen
	14
	15
	16	def _run(coro):
	17	asyncio.run(coro)
	18
	19
	20	def test_full_navigation(monkeypatch):
	21	def fake_run_audit(*, org, token, output_dir, branch, selected_keys, on_event):
	22	on_event(gh.ProgressEvent("start", "Member roster"))
	23	on_event(gh.ProgressEvent("done", "Member roster", count=3))
	24	on_event(gh.ProgressEvent("summary", output_dir, count=3))
	25	return [("Member roster", 3)]
	26
	27	monkeypatch.setattr("tui.github_runner.run_audit", fake_run_audit)
	28
	29	async def scenario():
	30	app = AuditApp()
	31	async with app.run_test(size=(120, 40)) as pilot:
	32	await pilot.pause()
	33	assert isinstance(app.screen, MenuScreen)
	34
	35	await pilot.click("#github")
	36	await pilot.pause()
	37	assert isinstance(app.screen, ConfigScreen)
	38
	39	app.screen.query_one("#org", Input).value = "acme"
	40	app.screen.query_one("#token", Input).value = "tok"
	41	await pilot.click("#continue")
	42	await pilot.pause()
	43	assert isinstance(app.screen, ChecksScreen)
	44	assert app.settings["org"] == "acme"
	45
	46	await pilot.click("#run")
	47	await pilot.pause()
	48	assert isinstance(app.screen, RunScreen)
	49
	50	await app.workers.wait_for_complete()
	51	await pilot.pause()
	52
	53	# When the run finishes, the exit buttons become enabled.
	54	assert app.screen.query_one("#menu", Button).disabled is False
	55	assert app.screen.query_one("#quit", Button).disabled is False
	56
	57	_run(scenario())
	58
	59
	60	def test_config_requires_org_and_token(monkeypatch):
	61	# Make sure env vars don't pre-fill the fields for this test.
	62	monkeypatch.delenv("GITHUB_ORG", raising=False)
	63	monkeypatch.delenv("GITHUB_TOKEN", raising=False)
	64
	65	async def scenario():
	66	app = AuditApp()
	67	async with app.run_test(size=(120, 40)) as pilot:
	68	await pilot.click("#github")
	69	await pilot.pause()
	70	# Continue with empty fields -> stays on ConfigScreen with an error.
	71	await pilot.click("#continue")
	72	await pilot.pause()
	73	assert isinstance(app.screen, ConfigScreen)
	74	error_text = str(app.screen.query_one("#form-error").render())
	75	assert "provide" in error_text.lower()
	76
	77	_run(scenario())
	78
	79
	80	def test_gitlab_is_enabled():
	81	async def scenario():
	82	app = AuditApp()
	83	async with app.run_test(size=(120, 40)) as pilot:
	84	await pilot.pause()
	85	assert app.screen.query_one("#gitlab", Button).disabled is False
	86
	87	_run(scenario())
	88
	89
	90	def test_gitlab_navigation(monkeypatch):
	91	monkeypatch.delenv("GITLAB_GROUP", raising=False)
	92	monkeypatch.delenv("GITLAB_TOKEN", raising=False)
	93
	94	def fake_run_audit(*, group, token, base_url, output_dir, selected_keys, on_event):
	95	on_event(gh.ProgressEvent("done", "Group members", count=7))
	96	on_event(gh.ProgressEvent("summary", output_dir, count=7))
	97	return [("Group members", 7)]
	98
	99	monkeypatch.setattr("tui.gitlab_runner.run_audit", fake_run_audit)
	100
	101	async def scenario():
	102	app = AuditApp()
	103	async with app.run_test(size=(120, 40)) as pilot:
	104	await pilot.pause()
	105	await pilot.click("#gitlab")
	106	await pilot.pause()
	107	assert isinstance(app.screen, ConfigScreen)
	108
	109	app.screen.query_one("#group", Input).value = "my-group"
	110	app.screen.query_one("#token", Input).value = "glpat-x"
	111	await pilot.click("#continue")
	112	await pilot.pause()
	113	assert isinstance(app.screen, ChecksScreen)
	114	assert app.settings["group"] == "my-group"
	115	# Self-hosted URL defaults to gitlab.com.
	116	assert app.settings["base_url"] == "https://gitlab.com/api/v4"
	117
	118	await pilot.click("#run")
	119	await pilot.pause()
	120	assert isinstance(app.screen, RunScreen)
	121	assert "gitlab_audit_my-group" in app.screen.output_dir
	122
	123	await app.workers.wait_for_complete()
	124	await pilot.pause()
	125	assert app.screen.query_one("#menu", Button).disabled is False
	126
	127	_run(scenario())

tui/tests/test_github_runner.py added +114

@@ -0,0 +1,114 @@
	1	"""Tests for the TUI's GitHub audit orchestration.
	2
	3	These stub out the network-bound collectors and verify run_audit's wiring:
	4	argument dispatch, the shared collaborator cache, per-check error handling,
	5	and that the CSV package (per-check files + summary) is written.
	6	"""
	7
	8	import csv
	9	import os
	10
	11	import pytest
	12
	13	from tui import github_runner as r
	14
	15
	16	@pytest.fixture
	17	def fake_checks(monkeypatch):
	18	"""Replace the real registry with stubbed collectors and record calls."""
	19	calls = {}
	20
	21	def base_fn(org, cfg):
	22	calls["base"] = (org, cfg)
	23	return [{"login": "alice"}, {"login": "bob"}]
	24
	25	def collabs_fn(org, cfg, repo_collabs):
	26	calls["collabs"] = (org, cfg, repo_collabs)
	27	return [{"repo": e["repo"]} for e in repo_collabs]
	28
	29	def branch_fn(org, cfg, branch):
	30	calls["branch"] = (org, cfg, branch)
	31	return [{"branch": branch}]
	32
	33	def boom_fn(org, cfg):
	34	raise RuntimeError("kaboom")
	35
	36	checks = [
	37	r.Check("base", "Base", base_fn, "base.csv"),
	38	r.Check("collabs", "Collabs", collabs_fn, "collabs.csv", arg="collabs"),
	39	r.Check("branch", "Branch", branch_fn, "branch.csv", arg="branch"),
	40	r.Check("boom", "Boom", boom_fn, "boom.csv"),
	41	]
	42	monkeypatch.setattr(r, "CHECKS", checks)
	43
	44	fetch_calls = []
	45
	46	def fake_fetch(org, cfg):
	47	fetch_calls.append(org)
	48	return [{"repo": "repo1", "collaborators": []}]
	49
	50	monkeypatch.setattr(r.members, "fetch_repo_collaborators", fake_fetch)
	51
	52	return calls, fetch_calls
	53
	54
	55	def run(tmp_path, keys, fake_checks, branch="main"):
	56	events = []
	57	sections = r.run_audit(
	58	org="acme",
	59	token="tok",
	60	output_dir=str(tmp_path),
	61	branch=branch,
	62	selected_keys=keys,
	63	on_event=events.append,
	64	)
	65	return events, sections
	66
	67
	68	def test_argument_dispatch_and_files(tmp_path, fake_checks):
	69	calls, _ = fake_checks
	70	run(tmp_path, ["base", "collabs", "branch"], fake_checks, "dev")
	71
	72	# Each collector was called with the right signature.
	73	assert calls["base"][0] == "acme"
	74	assert calls["collabs"][2] == [{"repo": "repo1", "collaborators": []}]
	75	assert calls["branch"][2] == "dev"
	76
	77	# CSV files were written for each check, plus the summary.
	78	for name in ("base.csv", "collabs.csv", "branch.csv", "summary.txt"):
	79	assert os.path.exists(tmp_path / name), name
	80
	81	with open(tmp_path / "base.csv", newline="") as f:
	82	assert len(list(csv.DictReader(f))) == 2
	83
	84
	85	def test_collab_cache_fetched_once(tmp_path, fake_checks):
	86	_, fetch_calls = fake_checks
	87	run(tmp_path, ["collabs", "base"], fake_checks)
	88	assert fetch_calls == ["acme"] # fetched exactly once
	89
	90
	91	def test_collab_cache_skipped_when_not_needed(tmp_path, fake_checks):
	92	_, fetch_calls = fake_checks
	93	run(tmp_path, ["base"], fake_checks)
	94	assert fetch_calls == [] # no collabs check selected -> no fetch
	95
	96
	97	def test_failing_check_does_not_abort_run(tmp_path, fake_checks):
	98	events, sections = run(tmp_path, ["boom", "base"], fake_checks)
	99
	100	kinds = [(e.kind, e.label) for e in events]
	101	assert ("error", "Boom") in kinds
	102	assert ("done", "Base") in kinds # base still ran after boom failed
	103
	104	labels = dict(sections)
	105	assert labels["Boom"] == 0
	106	assert labels["Base"] == 2
	107
	108
	109	def test_summary_event_totals_rows(tmp_path, fake_checks):
	110	events, _ = run(tmp_path, ["base", "branch"], fake_checks)
	111	summary = [e for e in events if e.kind == "summary"]
	112	assert len(summary) == 1
	113	assert summary[0].count == 3 # 2 base + 1 branch
	114	assert summary[0].label == str(tmp_path)

tui/tests/test_gitlab_runner.py added +113

@@ -0,0 +1,113 @@
	1	"""Tests for the TUI's GitLab audit orchestration.
	2
	3	Stub the network-bound collectors and verify run_audit's wiring: base vs
	4	project-scoped dispatch, the shared project cache, per-check error handling,
	5	and that the CSV package (per-check files + summary) is written.
	6	"""
	7
	8	import csv
	9	import os
	10
	11	import pytest
	12
	13	from tui import gitlab_runner as r
	14
	15
	16	@pytest.fixture
	17	def fake_checks(monkeypatch):
	18	calls = {}
	19
	20	def base_fn(group, cfg):
	21	calls["base"] = (group, cfg)
	22	return [{"username": "alice"}, {"username": "bob"}]
	23
	24	def projects_fn(group, cfg, projects):
	25	calls["projects"] = (group, cfg, projects)
	26	return [{"project": p["path_with_namespace"]} for p in projects]
	27
	28	def boom_fn(group, cfg):
	29	raise RuntimeError("kaboom")
	30
	31	checks = [
	32	r.Check("base", "Base", base_fn, "base.csv"),
	33	r.Check("projects", "Projects", projects_fn, "projects.csv", arg="projects"),
	34	r.Check("boom", "Boom", boom_fn, "boom.csv"),
	35	]
	36	monkeypatch.setattr(r, "CHECKS", checks)
	37
	38	fetch_calls = []
	39
	40	def fake_fetch(group, cfg):
	41	fetch_calls.append(group)
	42	return [{"id": 1, "path_with_namespace": "grp/proj"}]
	43
	44	monkeypatch.setattr(r.projects, "fetch_projects", fake_fetch)
	45
	46	return calls, fetch_calls
	47
	48
	49	def run(tmp_path, keys, base_url="https://gitlab.com/api/v4"):
	50	events = []
	51	sections = r.run_audit(
	52	group="grp",
	53	token="tok",
	54	base_url=base_url,
	55	output_dir=str(tmp_path),
	56	selected_keys=keys,
	57	on_event=events.append,
	58	)
	59	return events, sections
	60
	61
	62	def test_argument_dispatch_and_files(tmp_path, fake_checks):
	63	calls, _ = fake_checks
	64	run(tmp_path, ["base", "projects"])
	65
	66	assert calls["base"][0] == "grp"
	67	assert calls["projects"][2] == [{"id": 1, "path_with_namespace": "grp/proj"}]
	68
	69	for name in ("base.csv", "projects.csv", "summary.txt"):
	70	assert os.path.exists(tmp_path / name), name
	71
	72	with open(tmp_path / "projects.csv", newline="") as f:
	73	rows = list(csv.DictReader(f))
	74	assert rows == [{"project": "grp/proj"}]
	75
	76
	77	def test_project_cache_fetched_once(tmp_path, fake_checks):
	78	_, fetch_calls = fake_checks
	79	run(tmp_path, ["projects", "base"])
	80	assert fetch_calls == ["grp"]
	81
	82
	83	def test_project_cache_skipped_when_not_needed(tmp_path, fake_checks):
	84	_, fetch_calls = fake_checks
	85	run(tmp_path, ["base"])
	86	assert fetch_calls == []
	87
	88
	89	def test_failing_check_does_not_abort_run(tmp_path, fake_checks):
	90	events, sections = run(tmp_path, ["boom", "base"])
	91
	92	kinds = [(e.kind, e.label) for e in events]
	93	assert ("error", "Boom") in kinds
	94	assert ("done", "Base") in kinds
	95
	96	labels = dict(sections)
	97	assert labels["Boom"] == 0
	98	assert labels["Base"] == 2
	99
	100
	101	def test_base_url_reaches_cfg(tmp_path, fake_checks):
	102	calls, _ = fake_checks
	103	run(tmp_path, ["base"], base_url="https://gitlab.example.com/api/v4/")
	104	# build_cfg strips the trailing slash.
	105	assert calls["base"][1]["base_url"] == "https://gitlab.example.com/api/v4"
	106	assert calls["base"][1]["headers"]["PRIVATE-TOKEN"] == "tok"
	107
	108
	109	def test_premium_checks_off_by_default():
	110	assert "approval_rules" not in r.DEFAULT_SELECTION
	111	assert "audit_events" not in r.DEFAULT_SELECTION
	112	assert "password_policy" not in r.DEFAULT_SELECTION
	113	assert "group_members" in r.DEFAULT_SELECTION

Commit 31f6eb371d

Commit `31f6eb371d`