| 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 | ) |