1
+import asyncio
2
+from datetime import datetime
3
+from datetime import timezone
4
+from typing import Any
5
+from typing import Dict
6
+from typing import List
7
+from typing import Optional
8
+from typing import Tuple
9
+
10
+import httpx
11
+from loguru import logger
12
+
13
+from app.integrations.github_audit.schema.github_audit import AuditCheck
14
+from app.integrations.github_audit.schema.github_audit import AuditStatus
15
+from app.integrations.github_audit.schema.github_audit import AuditSummary
16
+from app.integrations.github_audit.schema.github_audit import GitHubAuditRequest
17
+from app.integrations.github_audit.schema.github_audit import GitHubAuditResponse
18
+from app.integrations.github_audit.schema.github_audit import GitHubAuditSummaryResponse
19
+from app.integrations.github_audit.schema.github_audit import MemberAuditResult
20
+from app.integrations.github_audit.schema.github_audit import OrganizationAuditResult
21
+from app.integrations.github_audit.schema.github_audit import RepositoryAuditResult
22
+from app.integrations.github_audit.schema.github_audit import SeverityLevel
23
+from app.integrations.github_audit.schema.github_audit import WorkflowAuditResult
24
+
25
+# GitHub API base URL
26
+GITHUB_API_BASE = "https://api.github.com"
27
+
28
+
29
+class GitHubAuditService:
30
+ """Service for performing GitHub organization security audits"""
31
+
32
+ def __init__(self, token: str, organization: str):
33
+ self.token = token
34
+ self.organization = organization
35
+ self.headers = {
36
+ "Authorization": f"Bearer {token}",
37
+ "Accept": "application/vnd.github+json",
38
+ "X-GitHub-Api-Version": "2022-11-28",
39
+ }
40
+ self.client: Optional[httpx.AsyncClient] = None
41
+
42
+ async def __aenter__(self):
43
+ self.client = httpx.AsyncClient(timeout=30.0)
44
+ return self
45
+
46
+ async def __aexit__(self, exc_type, exc_val, exc_tb):
47
+ if self.client:
48
+ await self.client.aclose()
49
+
50
+ async def _request(
51
+ self,
52
+ method: str,
53
+ endpoint: str,
54
+ params: Optional[Dict[str, Any]] = None,
55
+ ) -> Tuple[Optional[Dict[str, Any]], int]:
56
+ """Make a request to GitHub API"""
57
+ url = f"{GITHUB_API_BASE}{endpoint}"
58
+
59
+ try:
60
+ response = await self.client.request(
61
+ method,
62
+ url,
63
+ headers=self.headers,
64
+ params=params,
65
+ )
66
+ if response.status_code == 200:
67
+ return response.json(), response.status_code
68
+ elif response.status_code == 404:
69
+ return None, 404
70
+ else:
71
+ logger.warning(f"GitHub API error: {response.status_code} - {response.text[:200]}")
72
+ return None, response.status_code
73
+ except Exception as e:
74
+ logger.error(f"GitHub API request failed: {e}")
75
+ return None, 500
76
+
77
+ async def _paginate(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
78
+ """Paginate through GitHub API results"""
79
+ results = []
80
+ params = params or {}
81
+ params["per_page"] = 100
82
+ page = 1
83
+
84
+ while True:
85
+ params["page"] = page
86
+ data, status = await self._request("GET", endpoint, params)
87
+
88
+ if status != 200 or not data:
89
+ break
90
+
91
+ if isinstance(data, list):
92
+ results.extend(data)
93
+ if len(data) < 100:
94
+ break
95
+ else:
96
+ results.append(data)
97
+ break
98
+
99
+ page += 1
100
+
101
+ return results
102
+
103
+ # ==================== Organization Checks ====================
104
+
105
+ async def audit_organization(self) -> OrganizationAuditResult:
106
+ """Audit organization-level security settings"""
107
+ logger.info(f"Auditing organization: {self.organization}")
108
+
109
+ checks: List[AuditCheck] = []
110
+
111
+ # Get organization info
112
+ org_data, status = await self._request("GET", f"/orgs/{self.organization}")
113
+
114
+ if status != 200 or not org_data:
115
+ return OrganizationAuditResult(
116
+ org_name=self.organization,
117
+ org_url=f"https://github.com/{self.organization}",
118
+ checks=[
119
+ AuditCheck(
120
+ check_id="org-access",
121
+ check_name="Organization Access",
122
+ category="organization",
123
+ status=AuditStatus.FAIL,
124
+ severity=SeverityLevel.CRITICAL,
125
+ description="Unable to access organization",
126
+ recommendation="Verify the token has org:read permissions",
127
+ ),
128
+ ],
129
+ failed_count=1,
130
+ )
131
+
132
+ # Check: Two-factor authentication requirement
133
+ checks.append(await self._check_2fa_requirement(org_data))
134
+
135
+ # Check: Default repository permission
136
+ checks.append(await self._check_default_repo_permission(org_data))
137
+
138
+ # Check: Members can create repositories
139
+ checks.append(await self._check_member_repo_creation(org_data))
140
+
141
+ # Check: Members can create public repositories
142
+ checks.append(await self._check_public_repo_creation(org_data))
143
+
144
+ # Check: Verified domains
145
+ checks.append(await self._check_verified_domains())
146
+
147
+ # Check: SSO enforcement
148
+ checks.append(await self._check_sso_enforcement())
149
+
150
+ # Calculate counts
151
+ passed = sum(1 for c in checks if c.status == AuditStatus.PASS)
152
+ failed = sum(1 for c in checks if c.status == AuditStatus.FAIL)
153
+ warnings = sum(1 for c in checks if c.status == AuditStatus.WARNING)
154
+
155
+ return OrganizationAuditResult(
156
+ org_name=self.organization,
157
+ org_url=org_data.get("html_url", f"https://github.com/{self.organization}"),
158
+ checks=checks,
159
+ passed_count=passed,
160
+ failed_count=failed,
161
+ warning_count=warnings,
162
+ )
163
+
164
+ async def _check_2fa_requirement(self, org_data: Dict[str, Any]) -> AuditCheck:
165
+ """Check if 2FA is required for organization members"""
166
+ two_factor_required = org_data.get("two_factor_requirement_enabled", False)
167
+
168
+ return AuditCheck(
169
+ check_id="org-2fa-required",
170
+ check_name="Two-Factor Authentication Required",
171
+ category="organization",
172
+ status=AuditStatus.PASS if two_factor_required else AuditStatus.FAIL,
173
+ severity=SeverityLevel.CRITICAL,
174
+ description="Checks if 2FA is required for all organization members",
175
+ recommendation="Enable 2FA requirement in Organization Settings > Authentication security" if not two_factor_required else None,
176
+ resource_name=self.organization,
177
+ resource_type="organization",
178
+ )
179
+
180
+ async def _check_default_repo_permission(self, org_data: Dict[str, Any]) -> AuditCheck:
181
+ """Check default repository permission level"""
182
+ default_permission = org_data.get("default_repository_permission", "read")
183
+
184
+ # Acceptable: none, read. Risky: write, admin
185
+ is_secure = default_permission in ["none", "read"]
186
+
187
+ return AuditCheck(
188
+ check_id="org-default-permission",
189
+ check_name="Default Repository Permission",
190
+ category="organization",
191
+ status=AuditStatus.PASS if is_secure else AuditStatus.WARNING,
192
+ severity=SeverityLevel.MEDIUM,
193
+ description=f"Default repository permission is set to '{default_permission}'",
194
+ recommendation="Set default repository permission to 'read' or 'none' to follow least privilege" if not is_secure else None,
195
+ details={"current_permission": default_permission},
196
+ resource_name=self.organization,
197
+ resource_type="organization",
198
+ )
199
+
200
+ async def _check_member_repo_creation(self, org_data: Dict[str, Any]) -> AuditCheck:
201
+ """Check if members can create repositories"""
202
+ can_create = org_data.get("members_can_create_repositories", True)
203
+
204
+ return AuditCheck(
205
+ check_id="org-member-repo-creation",
206
+ check_name="Member Repository Creation",
207
+ category="organization",
208
+ status=AuditStatus.WARNING if can_create else AuditStatus.PASS,
209
+ severity=SeverityLevel.LOW,
210
+ description="Members can create repositories" if can_create else "Members cannot create repositories",
211
+ recommendation="Consider restricting repository creation to admins for better governance" if can_create else None,
212
+ resource_name=self.organization,
213
+ resource_type="organization",
214
+ )
215
+
216
+ async def _check_public_repo_creation(self, org_data: Dict[str, Any]) -> AuditCheck:
217
+ """Check if members can create public repositories"""
218
+ can_create_public = org_data.get("members_can_create_public_repositories", True)
219
+
220
+ return AuditCheck(
221
+ check_id="org-public-repo-creation",
222
+ check_name="Public Repository Creation",
223
+ category="organization",
224
+ status=AuditStatus.FAIL if can_create_public else AuditStatus.PASS,
225
+ severity=SeverityLevel.HIGH,
226
+ description="Members can create public repositories" if can_create_public else "Members cannot create public repositories",
227
+ recommendation="Restrict public repository creation to prevent accidental exposure of internal code"
228
+ if can_create_public
229
+ else None,
230
+ resource_name=self.organization,
231
+ resource_type="organization",
232
+ )
233
+
234
+ async def _check_verified_domains(self) -> AuditCheck:
235
+ """Check if organization has verified domains"""
236
+ domains, status = await self._request("GET", f"/orgs/{self.organization}/domains")
237
+
238
+ if status == 404:
239
+ return AuditCheck(
240
+ check_id="org-verified-domains",
241
+ check_name="Verified Domains",
242
+ category="organization",
243
+ status=AuditStatus.NOT_APPLICABLE,
244
+ severity=SeverityLevel.INFO,
245
+ description="Domain verification not available (requires GitHub Enterprise)",
246
+ resource_name=self.organization,
247
+ resource_type="organization",
248
+ )
249
+
250
+ verified_domains = []
251
+ if domains and isinstance(domains, list):
252
+ verified_domains = [d for d in domains if d.get("is_verified", False)]
253
+
254
+ has_verified = len(verified_domains) > 0
255
+
256
+ return AuditCheck(
257
+ check_id="org-verified-domains",
258
+ check_name="Verified Domains",
259
+ category="organization",
260
+ status=AuditStatus.PASS if has_verified else AuditStatus.WARNING,
261
+ severity=SeverityLevel.MEDIUM,
262
+ description=f"Organization has {len(verified_domains)} verified domain(s)"
263
+ if has_verified
264
+ else "Organization has no verified domains",
265
+ recommendation="Add and verify your organization's domain to improve trust and enable additional features"
266
+ if not has_verified
267
+ else None,
268
+ details={"verified_domains": [d.get("domain") for d in verified_domains]} if verified_domains else None,
269
+ resource_name=self.organization,
270
+ resource_type="organization",
271
+ )
272
+
273
+ async def _check_sso_enforcement(self) -> AuditCheck:
274
+ """Check if SAML SSO is enforced"""
275
+ # This requires enterprise API access
276
+ saml_data, status = await self._request("GET", f"/orgs/{self.organization}/saml")
277
+
278
+ if status == 404:
279
+ return AuditCheck(
280
+ check_id="org-sso-enforcement",
281
+ check_name="SAML SSO Enforcement",
282
+ category="organization",
283
+ status=AuditStatus.NOT_APPLICABLE,
284
+ severity=SeverityLevel.INFO,
285
+ description="SAML SSO configuration not available (requires GitHub Enterprise Cloud)",
286
+ resource_name=self.organization,
287
+ resource_type="organization",
288
+ )
289
+
290
+ enforced = saml_data.get("enforced", False) if saml_data else False
291
+
292
+ return AuditCheck(
293
+ check_id="org-sso-enforcement",
294
+ check_name="SAML SSO Enforcement",
295
+ category="organization",
296
+ status=AuditStatus.PASS if enforced else AuditStatus.WARNING,
297
+ severity=SeverityLevel.HIGH,
298
+ description="SAML SSO is enforced" if enforced else "SAML SSO is not enforced",
299
+ recommendation="Enable SAML SSO enforcement for centralized authentication" if not enforced else None,
300
+ resource_name=self.organization,
301
+ resource_type="organization",
302
+ )
303
+
304
+ # ==================== Repository Checks ====================
305
+
306
+ async def audit_repositories(
307
+ self,
308
+ repo_filter: Optional[List[str]] = None,
309
+ ) -> List[RepositoryAuditResult]:
310
+ """Audit all repositories in the organization"""
311
+ logger.info(f"Auditing repositories for organization: {self.organization}")
312
+
313
+ # Get all repositories
314
+ repos = await self._paginate(f"/orgs/{self.organization}/repos")
315
+
316
+ if not repos:
317
+ logger.warning("No repositories found or unable to fetch repositories")
318
+ return []
319
+
320
+ # Filter repos if specified
321
+ if repo_filter:
322
+ repos = [r for r in repos if r.get("name") in repo_filter]
323
+
324
+ results = []
325
+
326
+ # Audit repos concurrently in batches
327
+ batch_size = 10
328
+ for i in range(0, len(repos), batch_size):
329
+ batch = repos[i : i + batch_size]
330
+ batch_results = await asyncio.gather(*[self._audit_single_repo(repo) for repo in batch])
331
+ results.extend(batch_results)
332
+
333
+ return results
334
+
335
+ async def _audit_single_repo(self, repo: Dict[str, Any]) -> RepositoryAuditResult:
336
+ """Audit a single repository"""
337
+ repo_name = repo.get("name", "unknown")
338
+ full_name = repo.get("full_name", f"{self.organization}/{repo_name}")
339
+
340
+ logger.debug(f"Auditing repository: {full_name}")
341
+
342
+ checks: List[AuditCheck] = []
343
+
344
+ # Skip archived repos
345
+ if repo.get("archived", False):
346
+ return RepositoryAuditResult(
347
+ repo_name=repo_name,
348
+ repo_full_name=full_name,
349
+ repo_url=repo.get("html_url", ""),
350
+ is_private=repo.get("private", True),
351
+ is_archived=True,
352
+ default_branch=repo.get("default_branch", "main"),
353
+ checks=[
354
+ AuditCheck(
355
+ check_id="repo-archived",
356
+ check_name="Repository Archived",
357
+ category="repository",
358
+ status=AuditStatus.NOT_APPLICABLE,
359
+ severity=SeverityLevel.INFO,
360
+ description="Repository is archived - skipping security checks",
361
+ resource_name=repo_name,
362
+ resource_type="repository",
363
+ ),
364
+ ],
365
+ )
366
+
367
+ # Check: Branch protection
368
+ checks.append(await self._check_branch_protection(repo))
369
+
370
+ # Check: Secret scanning
371
+ checks.append(await self._check_secret_scanning(repo))
372
+
373
+ # Check: Dependabot alerts
374
+ checks.append(await self._check_dependabot_alerts(repo))
375
+
376
+ # Check: Code scanning
377
+ checks.append(await self._check_code_scanning(repo))
378
+
379
+ # Check: Private vulnerability reporting
380
+ checks.append(await self._check_private_vulnerability_reporting(repo))
381
+
382
+ # Check: License
383
+ checks.append(await self._check_license(repo))
384
+
385
+ # Check: Default branch protection
386
+ checks.append(await self._check_default_branch_deletion_protection(repo))
387
+
388
+ # Calculate counts
389
+ passed = sum(1 for c in checks if c.status == AuditStatus.PASS)
390
+ failed = sum(1 for c in checks if c.status == AuditStatus.FAIL)
391
+ warnings = sum(1 for c in checks if c.status == AuditStatus.WARNING)
392
+
393
+ return RepositoryAuditResult(
394
+ repo_name=repo_name,
395
+ repo_full_name=full_name,
396
+ repo_url=repo.get("html_url", ""),
397
+ is_private=repo.get("private", True),
398
+ is_archived=False,
399
+ default_branch=repo.get("default_branch", "main"),
400
+ checks=checks,
401
+ passed_count=passed,
402
+ failed_count=failed,
403
+ warning_count=warnings,
404
+ )
405
+
406
+ async def _check_branch_protection(self, repo: Dict[str, Any]) -> AuditCheck:
407
+ """Check if default branch has protection rules"""
408
+ repo_name = repo.get("name")
409
+ default_branch = repo.get("default_branch", "main")
410
+
411
+ protection, status = await self._request(
412
+ "GET",
413
+ f"/repos/{self.organization}/{repo_name}/branches/{default_branch}/protection",
414
+ )
415
+
416
+ if status == 404:
417
+ return AuditCheck(
418
+ check_id="repo-branch-protection",
419
+ check_name="Default Branch Protection",
420
+ category="repository",
421
+ status=AuditStatus.FAIL,
422
+ severity=SeverityLevel.HIGH,
423
+ description=f"Default branch '{default_branch}' has no protection rules",
424
+ recommendation="Enable branch protection rules to prevent direct pushes and require reviews",
425
+ resource_name=repo_name,
426
+ resource_type="repository",
427
+ )
428
+
429
+ # Check specific protection settings
430
+ details = {}
431
+ issues = []
432
+
433
+ if protection:
434
+ required_reviews = protection.get("required_pull_request_reviews")
435
+ if not required_reviews:
436
+ issues.append("Pull request reviews not required")
437
+ else:
438
+ details["required_approving_reviews"] = required_reviews.get("required_approving_review_count", 0)
439
+
440
+ if not protection.get("enforce_admins", {}).get("enabled", False):
441
+ issues.append("Admins can bypass protection")
442
+
443
+ if not protection.get("required_status_checks"):
444
+ issues.append("No required status checks")
445
+
446
+ details["dismiss_stale_reviews"] = (required_reviews or {}).get("dismiss_stale_reviews", False)
447
+ details["require_code_owner_reviews"] = (required_reviews or {}).get("require_code_owner_reviews", False)
448
+
449
+ if issues:
450
+ return AuditCheck(
451
+ check_id="repo-branch-protection",
452
+ check_name="Default Branch Protection",
453
+ category="repository",
454
+ status=AuditStatus.WARNING,
455
+ severity=SeverityLevel.MEDIUM,
456
+ description=f"Branch protection enabled but with gaps: {', '.join(issues)}",
457
+ recommendation="Strengthen branch protection by requiring reviews and enforcing for admins",
458
+ details=details,
459
+ resource_name=repo_name,
460
+ resource_type="repository",
461
+ )
462
+
463
+ return AuditCheck(
464
+ check_id="repo-branch-protection",
465
+ check_name="Default Branch Protection",
466
+ category="repository",
467
+ status=AuditStatus.PASS,
468
+ severity=SeverityLevel.HIGH,
469
+ description=f"Default branch '{default_branch}' has protection rules enabled",
470
+ details=details,
471
+ resource_name=repo_name,
472
+ resource_type="repository",
473
+ )
474
+
475
+ async def _check_secret_scanning(self, repo: Dict[str, Any]) -> AuditCheck:
476
+ """Check if secret scanning is enabled"""
477
+ repo_name = repo.get("name")
478
+
479
+ security_config, status = await self._request(
480
+ "GET",
481
+ f"/repos/{self.organization}/{repo_name}",
482
+ )
483
+
484
+ if status != 200:
485
+ return AuditCheck(
486
+ check_id="repo-secret-scanning",
487
+ check_name="Secret Scanning",
488
+ category="repository",
489
+ status=AuditStatus.NOT_APPLICABLE,
490
+ severity=SeverityLevel.INFO,
491
+ description="Unable to check secret scanning status",
492
+ resource_name=repo_name,
493
+ resource_type="repository",
494
+ )
495
+
496
+ security_and_analysis = security_config.get("security_and_analysis", {}) if security_config else {}
497
+ secret_scanning = security_and_analysis.get("secret_scanning", {})
498
+ secret_scanning_push = security_and_analysis.get("secret_scanning_push_protection", {})
499
+
500
+ scanning_enabled = secret_scanning.get("status") == "enabled"
501
+ push_protection_enabled = secret_scanning_push.get("status") == "enabled"
502
+
503
+ if scanning_enabled and push_protection_enabled:
504
+ return AuditCheck(
505
+ check_id="repo-secret-scanning",
506
+ check_name="Secret Scanning",
507
+ category="repository",
508
+ status=AuditStatus.PASS,
509
+ severity=SeverityLevel.HIGH,
510
+ description="Secret scanning and push protection are enabled",
511
+ details={"secret_scanning": True, "push_protection": True},
512
+ resource_name=repo_name,
513
+ resource_type="repository",
514
+ )
515
+ elif scanning_enabled:
516
+ return AuditCheck(
517
+ check_id="repo-secret-scanning",
518
+ check_name="Secret Scanning",
519
+ category="repository",
520
+ status=AuditStatus.WARNING,
521
+ severity=SeverityLevel.MEDIUM,
522
+ description="Secret scanning enabled but push protection is disabled",
523
+ recommendation="Enable secret scanning push protection to block secrets before they're committed",
524
+ details={"secret_scanning": True, "push_protection": False},
525
+ resource_name=repo_name,
526
+ resource_type="repository",
527
+ )
528
+ else:
529
+ return AuditCheck(
530
+ check_id="repo-secret-scanning",
531
+ check_name="Secret Scanning",
532
+ category="repository",
533
+ status=AuditStatus.FAIL,
534
+ severity=SeverityLevel.HIGH,
535
+ description="Secret scanning is not enabled",
536
+ recommendation="Enable secret scanning in repository Security settings",
537
+ details={"secret_scanning": False, "push_protection": False},
538
+ resource_name=repo_name,
539
+ resource_type="repository",
540
+ )
541
+
542
+ async def _check_dependabot_alerts(self, repo: Dict[str, Any]) -> AuditCheck:
543
+ """Check if Dependabot alerts are enabled"""
544
+ repo_name = repo.get("name")
545
+
546
+ # Check vulnerability alerts status
547
+ vuln_alerts, status = await self._request(
548
+ "GET",
549
+ f"/repos/{self.organization}/{repo_name}/vulnerability-alerts",
550
+ )
551
+
552
+ if status == 204: # 204 means enabled
553
+ return AuditCheck(
554
+ check_id="repo-dependabot-alerts",
555
+ check_name="Dependabot Alerts",
556
+ category="repository",
557
+ status=AuditStatus.PASS,
558
+ severity=SeverityLevel.HIGH,
559
+ description="Dependabot vulnerability alerts are enabled",
560
+ resource_name=repo_name,
561
+ resource_type="repository",
562
+ )
563
+ elif status == 404:
564
+ return AuditCheck(
565
+ check_id="repo-dependabot-alerts",
566
+ check_name="Dependabot Alerts",
567
+ category="repository",
568
+ status=AuditStatus.FAIL,
569
+ severity=SeverityLevel.HIGH,
570
+ description="Dependabot vulnerability alerts are not enabled",
571
+ recommendation="Enable Dependabot alerts in repository Security settings",
572
+ resource_name=repo_name,
573
+ resource_type="repository",
574
+ )
575
+ else:
576
+ return AuditCheck(
577
+ check_id="repo-dependabot-alerts",
578
+ check_name="Dependabot Alerts",
579
+ category="repository",
580
+ status=AuditStatus.NOT_APPLICABLE,
581
+ severity=SeverityLevel.INFO,
582
+ description="Unable to determine Dependabot alerts status",
583
+ resource_name=repo_name,
584
+ resource_type="repository",
585
+ )
586
+
587
+ async def _check_code_scanning(self, repo: Dict[str, Any]) -> AuditCheck:
588
+ """Check if code scanning is enabled"""
589
+ repo_name = repo.get("name")
590
+
591
+ # Check for code scanning alerts
592
+ alerts, status = await self._request(
593
+ "GET",
594
+ f"/repos/{self.organization}/{repo_name}/code-scanning/alerts",
595
+ params={"per_page": 1},
596
+ )
597
+
598
+ if status == 200:
599
+ return AuditCheck(
600
+ check_id="repo-code-scanning",
601
+ check_name="Code Scanning",
602
+ category="repository",
603
+ status=AuditStatus.PASS,
604
+ severity=SeverityLevel.MEDIUM,
605
+ description="Code scanning is enabled",
606
+ resource_name=repo_name,
607
+ resource_type="repository",
608
+ )
609
+ elif status == 404:
610
+ return AuditCheck(
611
+ check_id="repo-code-scanning",
612
+ check_name="Code Scanning",
613
+ category="repository",
614
+ status=AuditStatus.WARNING,
615
+ severity=SeverityLevel.MEDIUM,
616
+ description="Code scanning is not configured",
617
+ recommendation="Enable GitHub Advanced Security and configure CodeQL analysis",
618
+ resource_name=repo_name,
619
+ resource_type="repository",
620
+ )
621
+ else:
622
+ return AuditCheck(
623
+ check_id="repo-code-scanning",
624
+ check_name="Code Scanning",
625
+ category="repository",
626
+ status=AuditStatus.NOT_APPLICABLE,
627
+ severity=SeverityLevel.INFO,
628
+ description="Unable to determine code scanning status",
629
+ resource_name=repo_name,
630
+ resource_type="repository",
631
+ )
632
+
633
+ async def _check_private_vulnerability_reporting(self, repo: Dict[str, Any]) -> AuditCheck:
634
+ """Check if private vulnerability reporting is enabled"""
635
+ repo_name = repo.get("name")
636
+
637
+ # This is available in the repo data
638
+ pvr_enabled = repo.get("private_vulnerability_reporting_enabled", False)
639
+
640
+ return AuditCheck(
641
+ check_id="repo-private-vuln-reporting",
642
+ check_name="Private Vulnerability Reporting",
643
+ category="repository",
644
+ status=AuditStatus.PASS if pvr_enabled else AuditStatus.WARNING,
645
+ severity=SeverityLevel.LOW,
646
+ description="Private vulnerability reporting is enabled" if pvr_enabled else "Private vulnerability reporting is not enabled",
647
+ recommendation="Enable private vulnerability reporting to allow security researchers to report issues confidentially"
648
+ if not pvr_enabled
649
+ else None,
650
+ resource_name=repo_name,
651
+ resource_type="repository",
652
+ )
653
+
654
+ async def _check_license(self, repo: Dict[str, Any]) -> AuditCheck:
655
+ """Check if repository has a license"""
656
+ repo_name = repo.get("name")
657
+ license_info = repo.get("license")
658
+ is_private = repo.get("private", True)
659
+
660
+ if is_private:
661
+ return AuditCheck(
662
+ check_id="repo-license",
663
+ check_name="Repository License",
664
+ category="repository",
665
+ status=AuditStatus.NOT_APPLICABLE,
666
+ severity=SeverityLevel.INFO,
667
+ description="License check not applicable for private repositories",
668
+ resource_name=repo_name,
669
+ resource_type="repository",
670
+ )
671
+
672
+ has_license = license_info is not None
673
+
674
+ return AuditCheck(
675
+ check_id="repo-license",
676
+ check_name="Repository License",
677
+ category="repository",
678
+ status=AuditStatus.PASS if has_license else AuditStatus.WARNING,
679
+ severity=SeverityLevel.LOW,
680
+ description=f"Repository has license: {license_info.get('name', 'Unknown')}"
681
+ if has_license
682
+ else "Public repository has no license",
683
+ recommendation="Add a LICENSE file to clarify usage terms for public repositories" if not has_license else None,
684
+ details={"license": license_info.get("spdx_id") if license_info else None},
685
+ resource_name=repo_name,
686
+ resource_type="repository",
687
+ )
688
+
689
+ async def _check_default_branch_deletion_protection(self, repo: Dict[str, Any]) -> AuditCheck:
690
+ """Check if default branch deletion is protected"""
691
+ repo_name = repo.get("name")
692
+ default_branch = repo.get("default_branch", "main")
693
+
694
+ branch_info, status = await self._request(
695
+ "GET",
696
+ f"/repos/{self.organization}/{repo_name}/branches/{default_branch}",
697
+ )
698
+
699
+ if status != 200 or not branch_info:
700
+ return AuditCheck(
701
+ check_id="repo-branch-deletion",
702
+ check_name="Default Branch Deletion Protection",
703
+ category="repository",
704
+ status=AuditStatus.NOT_APPLICABLE,
705
+ severity=SeverityLevel.INFO,
706
+ description="Unable to check branch deletion protection",
707
+ resource_name=repo_name,
708
+ resource_type="repository",
709
+ )
710
+
711
+ protected = branch_info.get("protected", False)
712
+
713
+ return AuditCheck(
714
+ check_id="repo-branch-deletion",
715
+ check_name="Default Branch Deletion Protection",
716
+ category="repository",
717
+ status=AuditStatus.PASS if protected else AuditStatus.FAIL,
718
+ severity=SeverityLevel.HIGH,
719
+ description=f"Default branch '{default_branch}' is protected from deletion"
720
+ if protected
721
+ else f"Default branch '{default_branch}' can be deleted",
722
+ recommendation="Enable branch protection to prevent accidental deletion of default branch" if not protected else None,
723
+ resource_name=repo_name,
724
+ resource_type="repository",
725
+ )
726
+
727
+ # ==================== Workflow/Actions Checks ====================
728
+
729
+ async def audit_workflows(self) -> List[WorkflowAuditResult]:
730
+ """Audit GitHub Actions settings and workflows"""
731
+ logger.info(f"Auditing GitHub Actions for organization: {self.organization}")
732
+
733
+ results: List[WorkflowAuditResult] = []
734
+
735
+ # Get org-level actions permissions
736
+ actions_perms, status = await self._request("GET", f"/orgs/{self.organization}/actions/permissions")
737
+
738
+ if status == 200 and actions_perms:
739
+ checks = []
740
+
741
+ # Check allowed actions
742
+ allowed_actions = actions_perms.get("allowed_actions", "all")
743
+ if allowed_actions == "all":
744
+ checks.append(
745
+ AuditCheck(
746
+ check_id="actions-allowed-all",
747
+ check_name="Actions Permission Policy",
748
+ category="workflow",
749
+ status=AuditStatus.WARNING,
750
+ severity=SeverityLevel.MEDIUM,
751
+ description="All GitHub Actions are allowed to run",
752
+ recommendation="Restrict to verified creators or specific allowed actions",
753
+ resource_name=self.organization,
754
+ resource_type="organization",
755
+ ),
756
+ )
757
+ elif allowed_actions == "selected":
758
+ checks.append(
759
+ AuditCheck(
760
+ check_id="actions-allowed-selected",
761
+ check_name="Actions Permission Policy",
762
+ category="workflow",
763
+ status=AuditStatus.PASS,
764
+ severity=SeverityLevel.MEDIUM,
765
+ description="Only selected GitHub Actions are allowed",
766
+ resource_name=self.organization,
767
+ resource_type="organization",
768
+ ),
769
+ )
770
+
771
+ # Check default workflow permissions
772
+ default_perms, _ = await self._request("GET", f"/orgs/{self.organization}/actions/permissions/workflow")
773
+
774
+ if default_perms:
775
+ default_token_perms = default_perms.get("default_workflow_permissions", "write")
776
+ if default_token_perms == "write":
777
+ checks.append(
778
+ AuditCheck(
779
+ check_id="actions-default-token-perms",
780
+ check_name="Default Workflow Token Permissions",
781
+ category="workflow",
782
+ status=AuditStatus.WARNING,
783
+ severity=SeverityLevel.MEDIUM,
784
+ description="Default workflow token has write permissions",
785
+ recommendation="Set default workflow permissions to 'read' and grant write access explicitly where needed",
786
+ resource_name=self.organization,
787
+ resource_type="organization",
788
+ ),
789
+ )
790
+ else:
791
+ checks.append(
792
+ AuditCheck(
793
+ check_id="actions-default-token-perms",
794
+ check_name="Default Workflow Token Permissions",
795
+ category="workflow",
796
+ status=AuditStatus.PASS,
797
+ severity=SeverityLevel.MEDIUM,
798
+ description="Default workflow token has read-only permissions",
799
+ resource_name=self.organization,
800
+ resource_type="organization",
801
+ ),
802
+ )
803
+
804
+ results.append(
805
+ WorkflowAuditResult(
806
+ repo_name=self.organization,
807
+ workflow_name="Organization Actions Settings",
808
+ workflow_path="N/A",
809
+ checks=checks,
810
+ ),
811
+ )
812
+
813
+ return results
814
+
815
+ # ==================== Member Checks ====================
816
+
817
+ async def audit_members(self) -> List[MemberAuditResult]:
818
+ """Audit organization members"""
819
+ logger.info(f"Auditing members for organization: {self.organization}")
820
+
821
+ results: List[MemberAuditResult] = []
822
+
823
+ # Get all members
824
+ members = await self._paginate(f"/orgs/{self.organization}/members")
825
+
826
+ # Get admins
827
+ admins = await self._paginate(f"/orgs/{self.organization}/members", params={"role": "admin"})
828
+ admin_logins = set(m.get("login") for m in admins)
829
+
830
+ for member in members:
831
+ login = member.get("login", "unknown")
832
+ is_admin = login in admin_logins
833
+
834
+ checks = []
835
+
836
+ # Check: Admin count
837
+ if is_admin:
838
+ checks.append(
839
+ AuditCheck(
840
+ check_id="member-is-admin",
841
+ check_name="Admin Role",
842
+ category="member",
843
+ status=AuditStatus.WARNING,
844
+ severity=SeverityLevel.LOW,
845
+ description=f"User '{login}' has admin role",
846
+ recommendation="Regularly review admin access and remove if not needed",
847
+ resource_name=login,
848
+ resource_type="member",
849
+ ),
850
+ )
851
+
852
+ results.append(
853
+ MemberAuditResult(
854
+ username=login,
855
+ role="admin" if is_admin else "member",
856
+ checks=checks,
857
+ ),
858
+ )
859
+
860
+ return results
861
+
862
+ # ==================== Main Audit Function ====================
863
+
864
+ async def run_full_audit(self, request: GitHubAuditRequest) -> GitHubAuditResponse:
865
+ """Run a complete security audit"""
866
+ logger.info(f"Starting full security audit for organization: {self.organization}")
867
+
868
+ try:
869
+ org_results = await self.audit_organization()
870
+
871
+ repo_results = []
872
+ if request.include_repos:
873
+ repo_results = await self.audit_repositories(request.repo_filter)
874
+
875
+ workflow_results = []
876
+ if request.include_workflows:
877
+ workflow_results = await self.audit_workflows()
878
+
879
+ member_results = []
880
+ if request.include_members:
881
+ member_results = await self.audit_members()
882
+
883
+ # Calculate summary
884
+ all_checks: List[AuditCheck] = []
885
+ all_checks.extend(org_results.checks)
886
+ for repo in repo_results:
887
+ all_checks.extend(repo.checks)
888
+ for wf in workflow_results:
889
+ all_checks.extend(wf.checks)
890
+ for member in member_results:
891
+ all_checks.extend(member.checks)
892
+
893
+ # Filter out NOT_APPLICABLE checks for scoring
894
+ scorable_checks = [c for c in all_checks if c.status != AuditStatus.NOT_APPLICABLE]
895
+
896
+ total_checks = len(scorable_checks)
897
+ passed = sum(1 for c in scorable_checks if c.status == AuditStatus.PASS)
898
+ failed = sum(1 for c in scorable_checks if c.status == AuditStatus.FAIL)
899
+ warnings = sum(1 for c in scorable_checks if c.status == AuditStatus.WARNING)
900
+
901
+ # Count findings by severity (FAIL status only for severity counts)
902
+ critical = sum(1 for c in scorable_checks if c.status == AuditStatus.FAIL and c.severity == SeverityLevel.CRITICAL)
903
+ high = sum(1 for c in scorable_checks if c.status == AuditStatus.FAIL and c.severity == SeverityLevel.HIGH)
904
+ medium = sum(
905
+ 1 for c in scorable_checks if c.status in [AuditStatus.FAIL, AuditStatus.WARNING] and c.severity == SeverityLevel.MEDIUM
906
+ )
907
+ low = sum(1 for c in scorable_checks if c.status in [AuditStatus.FAIL, AuditStatus.WARNING] and c.severity == SeverityLevel.LOW)
908
+
909
+ # Calculate score using weighted pass rate
910
+ # Weight checks by severity: CRITICAL=4, HIGH=3, MEDIUM=2, LOW=1
911
+ if total_checks > 0:
912
+ total_weight = 0
913
+ earned_weight = 0
914
+
915
+ for check in scorable_checks:
916
+ weight = 1 # default
917
+ if check.severity == SeverityLevel.CRITICAL:
918
+ weight = 4
919
+ elif check.severity == SeverityLevel.HIGH:
920
+ weight = 3
921
+ elif check.severity == SeverityLevel.MEDIUM:
922
+ weight = 2
923
+ elif check.severity == SeverityLevel.LOW:
924
+ weight = 1
925
+
926
+ total_weight += weight
927
+ if check.status == AuditStatus.PASS:
928
+ earned_weight += weight
929
+ elif check.status == AuditStatus.WARNING:
930
+ # Warnings get partial credit
931
+ earned_weight += weight * 0.5
932
+
933
+ score = (earned_weight / total_weight) * 100 if total_weight > 0 else 100.0
934
+ else:
935
+ score = 100.0 # No checks = perfect score
936
+
937
+ # Round to 1 decimal place
938
+ score = round(score, 1)
939
+
940
+ # Determine grade
941
+ if score >= 90:
942
+ grade = "A"
943
+ elif score >= 80:
944
+ grade = "B"
945
+ elif score >= 70:
946
+ grade = "C"
947
+ elif score >= 60:
948
+ grade = "D"
949
+ else:
950
+ grade = "F"
951
+
952
+ logger.info(
953
+ f"Audit complete - Total: {total_checks}, Passed: {passed}, "
954
+ f"Failed: {failed}, Warnings: {warnings}, Score: {score}, Grade: {grade}",
955
+ )
956
+
957
+ summary = AuditSummary(
958
+ organization=self.organization,
959
+ audit_timestamp=datetime.now(timezone.utc).isoformat(),
960
+ total_repos_audited=len(repo_results),
961
+ total_checks=total_checks,
962
+ passed_checks=passed,
963
+ failed_checks=failed,
964
+ warning_checks=warnings,
965
+ critical_findings=critical,
966
+ high_findings=high,
967
+ medium_findings=medium,
968
+ low_findings=low,
969
+ score=score,
970
+ grade=grade,
971
+ )
972
+
973
+ # Get top findings (failed checks sorted by severity)
974
+ severity_order = {
975
+ SeverityLevel.CRITICAL: 0,
976
+ SeverityLevel.HIGH: 1,
977
+ SeverityLevel.MEDIUM: 2,
978
+ SeverityLevel.LOW: 3,
979
+ SeverityLevel.INFO: 4,
980
+ }
981
+ failed_checks = [c for c in all_checks if c.status in [AuditStatus.FAIL, AuditStatus.WARNING]]
982
+ top_findings = sorted(failed_checks, key=lambda c: severity_order.get(c.severity, 5))[:20]
983
+
984
+ return GitHubAuditResponse(
985
+ success=True,
986
+ message=f"Audit completed successfully. Score: {score} ({grade},)",
987
+ summary=summary,
988
+ organization_results=org_results,
989
+ repository_results=repo_results,
990
+ workflow_results=workflow_results,
991
+ member_results=member_results,
992
+ top_findings=top_findings,
993
+ )
994
+
995
+ except Exception as e:
996
+ logger.error(f"Audit failed: {e}")
997
+ return GitHubAuditResponse(
998
+ success=False,
999
+ message=f"Audit failed: {e}",
1000
+ summary=None,
1001
+ organization_results=None,
1002
+ repository_results=[],
1003
+ workflow_results=[],
1004
+ member_results=[],
1005
+ top_findings=[],
1006
+ )
1007
+
1008
+
1009
+# Service function wrappers
1010
+async def run_github_audit(
1011
+ token: str,
1012
+ request: GitHubAuditRequest,
1013
+) -> GitHubAuditResponse:
1014
+ """Run a GitHub organization security audit"""
1015
+ async with GitHubAuditService(token, request.organization) as service:
1016
+ return await service.run_full_audit(request)
1017
+
1018
+
1019
+async def run_github_audit_summary(
1020
+ token: str,
1021
+ request: GitHubAuditRequest,
1022
+) -> GitHubAuditSummaryResponse:
1023
+ """Run a GitHub audit and return summary only"""
1024
+ response = await run_github_audit(token, request)
1025
+
1026
+ return GitHubAuditSummaryResponse(
1027
+ success=response.success,
1028
+ message=response.message,
1029
+ summary=response.summary,
1030
+ top_findings=response.top_findings[:10],
1031
+ )