Download tests/scripts/emergency_coverage_bypass.py from techprotrade/annator-command-center: direct link, hf CLI and curl.
- Browser
- Download file 12.7 kB
-
https://huggingface.co/spaces/techprotrade/annator-command-center/resolve/main/tests/scripts/emergency_coverage_bypass.py
- Command line
-
hf download hf://spaces/techprotrade/annator-command-center/tests/scripts/emergency_coverage_bypass.py
-
curl -L -o emergency_coverage_bypass.py https://huggingface.co/spaces/techprotrade/annator-command-center/resolve/main/tests/scripts/emergency_coverage_bypass.py
12.7 kB
| #!/usr/bin/env python3 | |
| """ | |
| Emergency Coverage Bypass Tracking Script | |
| Purpose: Track and alert on emergency bypass usage to prevent abuse while allowing | |
| critical PRs (security fixes, hotfixes) to bypass coverage gates with approval. | |
| Usage: | |
| export EMERGENCY_COVERAGE_BYPASS=true | |
| export GITHUB_PR_URL="https://github.com/rushiparikh/atom/pull/1234" | |
| export BYPASS_REASON="Security fix: Critical authentication vulnerability" | |
| export GITHUB_APPROVERS="alice,bob" | |
| python emergency_coverage_bypass.py | |
| Features: | |
| - Logs bypass usage to JSON file with timestamp, reason, PR URL, approvers | |
| - Checks bypass frequency (>3 bypasses in 30 days triggers warning) | |
| - Sends alert notifications (Slack webhook placeholder) | |
| - Provides audit trail for monthly review process | |
| Output: | |
| - Console: Bypass status and alert messages | |
| - File: backend/tests/coverage_reports/metrics/bypass_log.json | |
| """ | |
| import json | |
| import os | |
| import sys | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Dict, List, Any | |
| # Path configuration | |
| BYPASS_LOG_PATH = Path(__file__).parent.parent / "coverage_reports" / "metrics" / "bypass_log.json" | |
| # Bypass frequency threshold (triggers investigation if exceeded) | |
| BYPASS_FREQUENCY_THRESHOLD = 3 | |
| BYPASS_FREQUENCY_WINDOW_DAYS = 30 | |
| def track_bypass_usage( | |
| reason: str, | |
| pr_url: str, | |
| approvers: List[str], | |
| phase: str, | |
| environment: str | |
| ) -> Dict[str, Any]: | |
| """ | |
| Track emergency bypass usage for audit trail. | |
| Args: | |
| reason: Bypass reason (e.g., "Security fix: Critical auth vulnerability") | |
| pr_url: Pull request URL (e.g., "https://github.com/rushiparikh/atom/pull/1234") | |
| approvers: List of approver usernames (e.g., ["alice", "bob"]) | |
| phase: Current coverage phase (e.g., "phase_1", "phase_2", "phase_3") | |
| environment: Environment (e.g., "production", "staging", "unknown") | |
| Returns: | |
| Dict containing bypass entry with timestamp, metadata | |
| """ | |
| # Create bypass entry | |
| entry = { | |
| "timestamp": datetime.now(timezone.utc).isoformat(), | |
| "reason": reason, | |
| "pr_url": pr_url, | |
| "approvers": approvers, | |
| "phase": phase, | |
| "environment": environment | |
| } | |
| # Load existing log | |
| if BYPASS_LOG_PATH.exists(): | |
| try: | |
| with open(BYPASS_LOG_PATH, 'r') as f: | |
| log = json.load(f) | |
| except (json.JSONDecodeError, IOError) as e: | |
| print(f"⚠️ Warning: Failed to load bypass log: {e}") | |
| log = {"bypasses": []} | |
| else: | |
| log = {"bypasses": []} | |
| # Add entry | |
| log["bypasses"].append(entry) | |
| # Save log | |
| BYPASS_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) | |
| try: | |
| with open(BYPASS_LOG_PATH, 'w') as f: | |
| json.dump(log, f, indent=2) | |
| except IOError as e: | |
| print(f"⚠️ Warning: Failed to save bypass log: {e}") | |
| return entry | |
| def check_bypass_frequency() -> bool: | |
| """ | |
| Check if bypass usage exceeds threshold (>3 per month). | |
| Returns: | |
| True if bypass frequency exceeds threshold (investigation needed) | |
| False if bypass usage within acceptable range | |
| """ | |
| if not BYPASS_LOG_PATH.exists(): | |
| return False | |
| try: | |
| with open(BYPASS_LOG_PATH, 'r') as f: | |
| log = json.load(f) | |
| except (json.JSONDecodeError, IOError): | |
| return False | |
| # Count bypasses in last 30 days | |
| cutoff_timestamp = datetime.now(timezone.utc).timestamp() - (BYPASS_FREQUENCY_WINDOW_DAYS * 24 * 60 * 60) | |
| recent_bypasses = [] | |
| for bypass in log.get("bypasses", []): | |
| try: | |
| bypass_time = datetime.fromisoformat(bypass["timestamp"].replace('Z', '+00:00')).timestamp() | |
| if bypass_time > cutoff_timestamp: | |
| recent_bypasses.append(bypass) | |
| except (ValueError, KeyError): | |
| # Skip invalid entries | |
| continue | |
| if len(recent_bypasses) > BYPASS_FREQUENCY_THRESHOLD: | |
| print(f"⚠️ WARNING: More than {BYPASS_FREQUENCY_THRESHOLD} emergency bypasses in last {BYPASS_FREQUENCY_WINDOW_DAYS} days") | |
| print(f" Recent bypasses: {len(recent_bypasses)}") | |
| print(f" Consider: Investigating root causes, adjusting thresholds") | |
| print(f" Recent bypass reasons:") | |
| for bypass in recent_bypasses[-5:]: # Show last 5 | |
| reason = bypass.get("reason", "No reason provided")[:60] | |
| date = bypass.get("timestamp", "")[:10] | |
| print(f" - {date}: {reason}") | |
| return True | |
| return False | |
| def check_bypass_eligibility(justification: str) -> bool: | |
| """ | |
| Check if emergency bypass is eligible for use with given justification. | |
| Args: | |
| justification: Required justification string for bypass (must be non-empty) | |
| Returns: | |
| True if bypass is allowed (valid justification and acceptable frequency) | |
| False if bypass is rejected (empty justification or excessive frequency) | |
| This function is called by backend_coverage_gate.py to determine if | |
| emergency bypass should be granted based on justification quality and | |
| recent bypass frequency. | |
| """ | |
| # Check 1: Justification must be provided and non-empty | |
| if not justification or not justification.strip(): | |
| print("❌ EMERGENCY BYPASS REJECTED: Justification is required") | |
| print(" Provide justification via BYPASS_REASON environment variable") | |
| return False | |
| # Check 2: Justification must have minimum length (prevent "test", "fix", etc.) | |
| if len(justification.strip()) < 20: | |
| print("❌ EMERGENCY BYPASS REJECTED: Justification too brief") | |
| print(" Justification must be at least 20 characters") | |
| print(" Example: 'Security fix: Critical auth vulnerability in production'") | |
| return False | |
| # Check 3: Bypass frequency check | |
| exceeds_threshold = check_bypass_frequency() | |
| # Log the bypass attempt for audit trail | |
| entry = track_bypass_usage( | |
| reason=justification, | |
| pr_url=os.getenv("GITHUB_PR_URL", "unknown"), | |
| approvers=[a.strip() for a in os.getenv("GITHUB_APPROVERS", "").split(",") if a.strip()] or ["unknown"], | |
| phase=os.getenv("COVERAGE_PHASE", "phase_1"), | |
| environment=os.getenv("ENVIRONMENT", "unknown") | |
| ) | |
| # Send alert notification | |
| send_bypass_alert(entry) | |
| # Even if frequency exceeds threshold, we allow the bypass but warn | |
| if exceeds_threshold: | |
| print("⚠️ BYPASS GRANTED WITH WARNING: Frequent bypass usage detected") | |
| print(" Please investigate root causes to avoid future bypasses") | |
| return True | |
| print("✅ EMERGENCY BYPASS GRANTED: Valid justification provided") | |
| return True | |
| def send_bypass_alert(entry: Dict): | |
| """ | |
| Send alert notification for bypass usage. | |
| Current implementation: Console output (placeholder for Slack webhook) | |
| Future: Integrate with Slack webhook for team notification | |
| Args: | |
| entry: Bypass entry dict with timestamp, reason, pr_url, approvers, phase | |
| """ | |
| print("🚨 EMERGENCY COVERAGE BYPASS ACTIVATED") | |
| print(f" Reason: {entry['reason']}") | |
| print(f" PR: {entry['pr_url']}") | |
| print(f" Approvers: {', '.join(entry['approvers'])}") | |
| print(f" Phase: {entry['phase']}") | |
| print(f" Environment: {entry['environment']}") | |
| print(f" Timestamp: {entry['timestamp']}") | |
| print() | |
| # Future: Integrate with Slack webhook | |
| # webhook_url = os.getenv("SLACK_COVERAGE_WEBHOOK") | |
| # if webhook_url: | |
| # try: | |
| # import requests | |
| # message = { | |
| # "text": f"🚨 Coverage bypass: {entry['pr_url']}", | |
| # "attachments": [{ | |
| # "color": "warning", | |
| # "fields": [ | |
| # {"title": "Reason", "value": entry['reason']}, | |
| # {"title": "Approvers", "value": ', '.join(entry['approvers'])}, | |
| # {"title": "Phase", "value": entry['phase']} | |
| # ] | |
| # }] | |
| # } | |
| # requests.post(webhook_url, json=message) | |
| # except Exception as e: | |
| # print(f"⚠️ Warning: Failed to send Slack alert: {e}") | |
| def print_bypass_summary(): | |
| """Print summary of all bypasses in log.""" | |
| if not BYPASS_LOG_PATH.exists(): | |
| return | |
| try: | |
| with open(BYPASS_LOG_PATH, 'r') as f: | |
| log = json.load(f) | |
| except (json.JSONDecodeError, IOError): | |
| return | |
| total_bypasses = len(log.get("bypasses", [])) | |
| if total_bypasses == 0: | |
| return | |
| print(f"📊 Bypass Summary: {total_bypasses} total bypasses logged") | |
| print() | |
| # Show recent bypasses (last 5) | |
| recent = log.get("bypasses", [])[-5:] | |
| for i, bypass in enumerate(reversed(recent), 1): | |
| reason = bypass.get("reason", "No reason")[:50] | |
| pr = bypass.get("pr_url", "Unknown PR") | |
| print(f" {i}. {reason}") | |
| print(f" {pr}") | |
| print() | |
| def main(): | |
| """Main bypass tracking logic.""" | |
| import argparse | |
| parser = argparse.ArgumentParser(description="Emergency coverage bypass tracking") | |
| parser.add_argument( | |
| "--test", | |
| action="store_true", | |
| help="Run test mode to verify bypass behavior without logging" | |
| ) | |
| args = parser.parse_args() | |
| # Test mode: verify bypass functionality without logging | |
| if args.test: | |
| print("=== EMERGENCY BYPASS TEST MODE ===") | |
| print() | |
| # Test 1: Empty justification should fail | |
| print("Test 1: Empty justification") | |
| result = check_bypass_eligibility("") | |
| print(f" Result: {'PASS' if not result else 'FAIL'} (should reject empty justification)") | |
| print() | |
| # Test 2: Short justification should fail | |
| print("Test 2: Short justification (< 20 chars)") | |
| result = check_bypass_eligibility("test fix") | |
| print(f" Result: {'PASS' if not result else 'FAIL'} (should reject short justification)") | |
| print() | |
| # Test 3: Valid justification should pass | |
| print("Test 3: Valid justification (>= 20 chars)") | |
| # Clear the log temporarily for clean test | |
| if BYPASS_LOG_PATH.exists(): | |
| import shutil | |
| backup = BYPASS_LOG_PATH.with_suffix('.json.bak') | |
| shutil.copy(BYPASS_LOG_PATH, backup) | |
| try: | |
| result = check_bypass_eligibility("Security fix: Critical authentication vulnerability affecting production") | |
| print(f" Result: {'PASS' if result else 'FAIL'} (should accept valid justification)") | |
| # Restore backup to avoid polluting log | |
| shutil.move(backup, BYPASS_LOG_PATH) | |
| except Exception as e: | |
| print(f" Test error: {e}") | |
| if backup.exists(): | |
| shutil.move(backup, BYPASS_LOG_PATH) | |
| else: | |
| result = check_bypass_eligibility("Security fix: Critical authentication vulnerability affecting production") | |
| print(f" Result: {'PASS' if result else 'FAIL'} (should accept valid justification)") | |
| print() | |
| print("=== TEST MODE COMPLETE ===") | |
| print("All bypass functionality verified:") | |
| print(" ✓ Rejects empty justification") | |
| print(" ✓ Rejects short justification (< 20 chars)") | |
| print(" ✓ Accepts valid justification (>= 20 chars)") | |
| print(" ✓ Logs bypass events to audit trail") | |
| print(" ✓ Tracks bypass frequency") | |
| return 0 | |
| # Normal mode: check environment variable | |
| bypass_active = os.getenv("EMERGENCY_COVERAGE_BYPASS", "false").lower() == "true" | |
| if not bypass_active: | |
| print("✅ Coverage gate active (no bypass)") | |
| print() | |
| print_bypass_summary() | |
| return 0 | |
| print("⚠️ COVERAGE GATE BYPASSED (emergency mode)") | |
| print() | |
| # Get bypass metadata from environment | |
| pr_url = os.getenv("GITHUB_PR_URL", "unknown") | |
| reason = os.getenv("BYPASS_REASON", "not provided") | |
| approvers_str = os.getenv("GITHUB_APPROVERS", "") | |
| approvers = [a.strip() for a in approvers_str.split(",") if a.strip()] if approvers_str else ["unknown"] | |
| phase = os.getenv("COVERAGE_PHASE", "phase_1") | |
| environment = os.getenv("ENVIRONMENT", "unknown") | |
| # Track usage | |
| entry = track_bypass_usage(reason, pr_url, approvers, phase, environment) | |
| # Send alert | |
| send_bypass_alert(entry) | |
| # Check frequency | |
| check_bypass_frequency() | |
| print() | |
| print("📝 Bypass logged to:", BYPASS_LOG_PATH) | |
| print("⚠️ Remember to remove EMERGENCY_COVERAGE_BYPASS after PR merges") | |
| print() | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |