From acd8b10630449763a9be241f2e1a0e956ac394f1 Mon Sep 17 00:00:00 2001 From: krishna fattepurkar Date: Thu, 1 Jan 2026 19:24:20 +0530 Subject: [PATCH] chore(format): apply black code formatter to all Python files - Run black --line-length=100 on all source and test files - Reformatted 20 files to match Black formatting standards - 1 file left unchanged (already compliant) - Fixes CI/CD black formatting check failures - Ensures code style consistency across project This aligns all code with Black's opinionated formatting style, making the codebase consistent and passing CI/CD checks. --- src/shieldcommit/__main__.py | 17 +- src/shieldcommit/aks_detector.py | 91 ++-- src/shieldcommit/azure_db_detector.py | 143 +++--- src/shieldcommit/eks_detector.py | 91 ++-- src/shieldcommit/gcp_db_detector.py | 114 +++-- src/shieldcommit/gcp_detector.py | 130 ++--- src/shieldcommit/installer.py | 12 +- src/shieldcommit/intelligent_detector.py | 606 ++++++++++++++--------- src/shieldcommit/rds_detector.py | 101 ++-- src/shieldcommit/scanner.py | 13 +- src/shieldcommit/utils.py | 2 + tests/test_aks_detector.py | 60 ++- tests/test_azure_db_detector.py | 80 +-- tests/test_eks_detector.py | 72 +-- tests/test_gcp_db_detector.py | 90 ++-- tests/test_gcp_detector.py | 70 +-- tests/test_intelligent_detector.py | 124 ++--- tests/test_patterns.py | 11 +- tests/test_rds_detector.py | 74 +-- tests/test_scanner.py | 43 +- 20 files changed, 1118 insertions(+), 826 deletions(-) diff --git a/src/shieldcommit/__main__.py b/src/shieldcommit/__main__.py index 74b1534..4941af0 100644 --- a/src/shieldcommit/__main__.py +++ b/src/shieldcommit/__main__.py @@ -5,6 +5,7 @@ from .scanner import scan_files from .installer import install_hook, uninstall_hook + def get_staged_files(): """Return list of staged files' paths.""" res = subprocess.run(["git", "diff", "--cached", "--name-only"], capture_output=True, text=True) @@ -15,11 +16,13 @@ def get_staged_files(): return [] return [p for p in out.splitlines() if p.strip()] + @click.group() def cli(): """ShieldCommit - Secret Scanner for Git Commits""" pass + @cli.command() @click.argument("paths", nargs=-1, type=click.Path(exists=False)) def scan(paths): @@ -45,13 +48,15 @@ def scan(paths): else: to_scan = get_staged_files() if not to_scan: - click.echo("No staged files. Use `shieldcommit scan ` to scan files or set staged files.") + click.echo( + "No staged files. Use `shieldcommit scan ` to scan files or set staged files." + ) sys.exit(0) result = scan_files(to_scan) findings = result["findings"] warnings = result["warnings"] - + # Display warnings (non-blocking) if warnings: click.echo("⚠️ VERSION WARNINGS (Info only - no block):\n") @@ -60,7 +65,7 @@ def scan(paths): click.echo(f" {w['message']}") click.echo(f" Snippet: {w['snippet']}") click.echo("") - + # Display findings (blocking) if not findings: click.echo("✓ No secrets found.") @@ -73,9 +78,12 @@ def scan(paths): click.echo(f" Confidence: {f.get('confidence', 'N/A'):.2%}") click.echo(f" Snippet: {f['snippet']}") click.echo("") - click.echo("Your commit or action has been blocked. Remove or rotate secrets before proceeding.") + click.echo( + "Your commit or action has been blocked. Remove or rotate secrets before proceeding." + ) sys.exit(1) + @cli.command() def install(): """Install pre-commit hook in current repo.""" @@ -85,6 +93,7 @@ def install(): else: click.echo("❌ Failed to install hook.") + @cli.command() def uninstall(): ok = uninstall_hook(".") diff --git a/src/shieldcommit/aks_detector.py b/src/shieldcommit/aks_detector.py index c6b8c61..26ffb6a 100644 --- a/src/shieldcommit/aks_detector.py +++ b/src/shieldcommit/aks_detector.py @@ -21,6 +21,7 @@ "1.22": {"status": "deprecated", "eol": "2023-11"}, } + def parse_terraform_aks_versions(content: str) -> List[Dict[str, Any]]: """ Parse Terraform file content for AKS cluster versions. @@ -31,95 +32,105 @@ def parse_terraform_aks_versions(content: str) -> List[Dict[str, Any]]: """ findings = [] lines = content.splitlines() - + # Pattern 1: Direct resource usage: kubernetes_version = "1.27" pattern1 = r'kubernetes_version\s*=\s*["\']([0-9]+\.[0-9]+)["\']' - + # Pattern 2: Variable default values: default = "1.27" (for cluster_version or similar) pattern2 = r'default\s*=\s*["\']([0-9]+\.[0-9]+)["\']' - + in_cluster_version_var = False - + for line_no, line in enumerate(lines, 1): # Track if we're in a cluster_version variable block - if 'variable' in line and 'cluster_version' in line: + if "variable" in line and "cluster_version" in line: in_cluster_version_var = True - elif in_cluster_version_var and '}' in line: + elif in_cluster_version_var and "}" in line: in_cluster_version_var = False - + # Check pattern 1: Direct kubernetes_version matches = re.finditer(pattern1, line) for match in matches: version = match.group(1) - findings.append({ - "line": line_no, - "version": version, - "snippet": line.strip(), - "match": match.group(0) - }) - + findings.append( + { + "line": line_no, + "version": version, + "snippet": line.strip(), + "match": match.group(0), + } + ) + # Check pattern 2: Variable default (when in cluster_version variable block) if in_cluster_version_var: matches = re.finditer(pattern2, line) for match in matches: version = match.group(1) - findings.append({ - "line": line_no, - "version": version, - "snippet": line.strip(), - "match": match.group(0) - }) - + findings.append( + { + "line": line_no, + "version": version, + "snippet": line.strip(), + "match": match.group(0), + } + ) + return findings + def get_version_warning(version: str) -> str: """Get warning message for a given AKS version.""" if version not in AKS_VERSIONS: return f"⚠️ AKS version {version} is unknown. Check Azure documentation." - + info = AKS_VERSIONS[version] status = info["status"] eol = info["eol"] - + if status == "deprecated": return f"🚨 AKS {version} is DEPRECATED (EOL: {eol}). Upgrade immediately." elif status == "extended": - return f"⚠️ AKS {version} on Extended Support (EOL: {eol}). Higher costs. Consider upgrading." + return ( + f"⚠️ AKS {version} on Extended Support (EOL: {eol}). Higher costs. Consider upgrading." + ) else: # current return f"✓ AKS {version} is currently supported (EOL: {eol})." + def scan_aks_versions(file_path: Path) -> List[Dict[str, Any]]: """ Scan a Terraform file for AKS version warnings. Returns list of warnings. """ warnings = [] - + # Only scan Terraform files - if file_path.suffix not in {'.tf', '.json'}: + if file_path.suffix not in {".tf", ".json"}: return warnings - + try: content = file_path.read_text(errors="ignore") except Exception: return warnings - + findings = parse_terraform_aks_versions(content) - + for finding in findings: version = finding["version"] warning_msg = get_version_warning(version) - + # Only warn on non-current versions if version in AKS_VERSIONS and AKS_VERSIONS[version]["status"] != "current": - warnings.append({ - "file": str(file_path), - "line": finding["line"], - "type": "aks_version", - "version": version, - "status": AKS_VERSIONS[version]["status"], - "message": warning_msg, - "snippet": finding["snippet"] - }) - + warnings.append( + { + "file": str(file_path), + "line": finding["line"], + "type": "aks_version", + "version": version, + "status": AKS_VERSIONS[version]["status"], + "message": warning_msg, + "snippet": finding["snippet"], + } + ) + return warnings diff --git a/src/shieldcommit/azure_db_detector.py b/src/shieldcommit/azure_db_detector.py index b3f04eb..7ccd7b6 100644 --- a/src/shieldcommit/azure_db_detector.py +++ b/src/shieldcommit/azure_db_detector.py @@ -22,7 +22,7 @@ "2017": {"status": "deprecated", "eol": "2024-10"}, "2016": {"status": "deprecated", "eol": "2024-07"}, "2014": {"status": "deprecated", "eol": "2024-07"}, - } + }, }, # Azure MySQL versions "mysql": { @@ -31,7 +31,7 @@ "8.0": {"status": "current", "eol": "2026-04"}, "5.7": {"status": "deprecated", "eol": "2024-10"}, "5.6": {"status": "deprecated", "eol": "2021-02"}, - } + }, }, # Azure PostgreSQL versions "postgres": { @@ -43,10 +43,11 @@ "13": {"status": "extended", "eol": "2025-11"}, "12": {"status": "deprecated", "eol": "2024-11"}, "11": {"status": "deprecated", "eol": "2023-10"}, - } + }, }, } + def parse_terraform_azure_db_versions(content: str) -> List[Dict[str, Any]]: """ Parse Terraform file content for Azure database versions. @@ -57,115 +58,125 @@ def parse_terraform_azure_db_versions(content: str) -> List[Dict[str, Any]]: """ findings = [] lines = content.splitlines() - + # Pattern for SQL Server sql_pattern = r'sku_name\s*=\s*["\']([A-Z0-9]+)["\']' # Pattern for MySQL server version mysql_pattern = r'version\s*=\s*["\']([0-9.]+)["\']' # Pattern for PostgreSQL server version postgres_pattern = r'version\s*=\s*["\']([0-9.]+)["\']' - + current_resource = None - + for line_no, line in enumerate(lines, 1): # Detect resource type - if 'azurerm_mssql_server' in line: - current_resource = 'sql' - elif 'azurerm_mysql_server' in line or 'azurerm_mysql_flexible_server' in line: - current_resource = 'mysql' - elif 'azurerm_postgresql_server' in line or 'azurerm_postgresql_flexible_server' in line: - current_resource = 'postgres' - + if "azurerm_mssql_server" in line: + current_resource = "sql" + elif "azurerm_mysql_server" in line or "azurerm_mysql_flexible_server" in line: + current_resource = "mysql" + elif "azurerm_postgresql_server" in line or "azurerm_postgresql_flexible_server" in line: + current_resource = "postgres" + # Match SQL versions - if current_resource == 'sql': + if current_resource == "sql": sql_matches = re.finditer(sql_pattern, line) for match in sql_matches: sku = match.group(1) # Extract SQL Server version from SKU (e.g., GP_Gen5_2 -> infer from context) # For simplicity, we'll note the SKU - findings.append({ - "line": line_no, - "engine": "sql", - "version": sku, - "snippet": line.strip(), - "match": match.group(0), - "type": "sku" - }) - + findings.append( + { + "line": line_no, + "engine": "sql", + "version": sku, + "snippet": line.strip(), + "match": match.group(0), + "type": "sku", + } + ) + # Match MySQL versions - elif current_resource == 'mysql': + elif current_resource == "mysql": mysql_matches = re.finditer(mysql_pattern, line) for match in mysql_matches: version = match.group(1) - findings.append({ - "line": line_no, - "engine": "mysql", - "version": version, - "snippet": line.strip(), - "match": match.group(0), - "type": "version" - }) - + findings.append( + { + "line": line_no, + "engine": "mysql", + "version": version, + "snippet": line.strip(), + "match": match.group(0), + "type": "version", + } + ) + # Match PostgreSQL versions - elif current_resource == 'postgres': + elif current_resource == "postgres": postgres_matches = re.finditer(postgres_pattern, line) for match in postgres_matches: version = match.group(1) - findings.append({ - "line": line_no, - "engine": "postgres", - "version": version, - "snippet": line.strip(), - "match": match.group(0), - "type": "version" - }) - + findings.append( + { + "line": line_no, + "engine": "postgres", + "version": version, + "snippet": line.strip(), + "match": match.group(0), + "type": "version", + } + ) + return findings + def get_azure_db_warning(engine: str, version: str) -> str: """Get warning message for a given Azure database version.""" if engine not in AZURE_DB_ENGINES: return f"⚠️ Azure database engine '{engine}' is not recognized. Check Azure documentation." - + engine_info = AZURE_DB_ENGINES[engine] versions = engine_info.get("versions", {}) - + if version not in versions: return f"⚠️ {engine.upper()} version {version} is unknown. Check Azure documentation." - + info = versions[version] status = info["status"] eol = info["eol"] - + if status == "deprecated": - return f"🚨 Azure {engine.upper()} {version} is DEPRECATED (EOL: {eol}). Upgrade immediately." + return ( + f"🚨 Azure {engine.upper()} {version} is DEPRECATED (EOL: {eol}). Upgrade immediately." + ) elif status == "extended": return f"⚠️ Azure {engine.upper()} {version} on Extended Support (EOL: {eol}). Higher costs. Consider upgrading." else: # current return f"✓ Azure {engine.upper()} {version} is currently supported (EOL: {eol})." + def scan_azure_db_versions(file_path: Path) -> List[Dict[str, Any]]: """ Scan a Terraform file for Azure database version warnings. Returns list of warnings. """ warnings = [] - + # Only scan Terraform files - if file_path.suffix not in {'.tf', '.json'}: + if file_path.suffix not in {".tf", ".json"}: return warnings - + try: content = file_path.read_text(errors="ignore") except Exception: return warnings - + findings = parse_terraform_azure_db_versions(content) - + for finding in findings: engine = finding["engine"] version = finding["version"] - + if engine in AZURE_DB_ENGINES: versions = AZURE_DB_ENGINES[engine].get("versions", {}) if version in versions: @@ -173,15 +184,17 @@ def scan_azure_db_versions(file_path: Path) -> List[Dict[str, Any]]: # Only warn on non-current versions if info["status"] != "current": warning_msg = get_azure_db_warning(engine, version) - warnings.append({ - "file": str(file_path), - "line": finding["line"], - "type": "azure_db_version", - "engine": engine, - "version": version, - "status": info["status"], - "message": warning_msg, - "snippet": finding["snippet"] - }) - + warnings.append( + { + "file": str(file_path), + "line": finding["line"], + "type": "azure_db_version", + "engine": engine, + "version": version, + "status": info["status"], + "message": warning_msg, + "snippet": finding["snippet"], + } + ) + return warnings diff --git a/src/shieldcommit/eks_detector.py b/src/shieldcommit/eks_detector.py index f61a650..1f8b962 100644 --- a/src/shieldcommit/eks_detector.py +++ b/src/shieldcommit/eks_detector.py @@ -22,6 +22,7 @@ "1.21": {"status": "deprecated", "eol": "2023-04"}, } + def parse_terraform_eks_versions(content: str) -> List[Dict[str, Any]]: """ Parse Terraform file content for EKS cluster versions. @@ -32,95 +33,105 @@ def parse_terraform_eks_versions(content: str) -> List[Dict[str, Any]]: """ findings = [] lines = content.splitlines() - + # Pattern 1: Direct resource usage: kubernetes_version = "1.27" pattern1 = r'kubernetes_version\s*=\s*["\']([0-9]+\.[0-9]+)["\']' - + # Pattern 2: Variable default values: default = "1.27" (for cluster_version or similar) pattern2 = r'default\s*=\s*["\']([0-9]+\.[0-9]+)["\']' - + in_cluster_version_var = False - + for line_no, line in enumerate(lines, 1): # Track if we're in a cluster_version variable block - if 'variable' in line and 'cluster_version' in line: + if "variable" in line and "cluster_version" in line: in_cluster_version_var = True - elif in_cluster_version_var and '}' in line: + elif in_cluster_version_var and "}" in line: in_cluster_version_var = False - + # Check pattern 1: Direct kubernetes_version matches = re.finditer(pattern1, line) for match in matches: version = match.group(1) - findings.append({ - "line": line_no, - "version": version, - "snippet": line.strip(), - "match": match.group(0) - }) - + findings.append( + { + "line": line_no, + "version": version, + "snippet": line.strip(), + "match": match.group(0), + } + ) + # Check pattern 2: Variable default (when in cluster_version variable block) if in_cluster_version_var: matches = re.finditer(pattern2, line) for match in matches: version = match.group(1) - findings.append({ - "line": line_no, - "version": version, - "snippet": line.strip(), - "match": match.group(0) - }) - + findings.append( + { + "line": line_no, + "version": version, + "snippet": line.strip(), + "match": match.group(0), + } + ) + return findings + def get_version_warning(version: str) -> str: """Get warning message for a given EKS version.""" if version not in EKS_VERSIONS: return f"⚠️ EKS version {version} is unknown. Check AWS documentation." - + info = EKS_VERSIONS[version] status = info["status"] eol = info["eol"] - + if status == "deprecated": return f"🚨 EKS {version} is DEPRECATED (EOL: {eol}). Upgrade immediately." elif status == "extended": - return f"⚠️ EKS {version} on Extended Support (EOL: {eol}). Higher costs. Consider upgrading." + return ( + f"⚠️ EKS {version} on Extended Support (EOL: {eol}). Higher costs. Consider upgrading." + ) else: # current return f"✓ EKS {version} is currently supported (EOL: {eol})." + def scan_eks_versions(file_path: Path) -> List[Dict[str, Any]]: """ Scan a Terraform file for EKS version warnings. Returns list of warnings. """ warnings = [] - + # Only scan Terraform files - if file_path.suffix not in {'.tf', '.json'}: + if file_path.suffix not in {".tf", ".json"}: return warnings - + try: content = file_path.read_text(errors="ignore") except Exception: return warnings - + findings = parse_terraform_eks_versions(content) - + for finding in findings: version = finding["version"] warning_msg = get_version_warning(version) - + # Only warn on non-current versions if version in EKS_VERSIONS and EKS_VERSIONS[version]["status"] != "current": - warnings.append({ - "file": str(file_path), - "line": finding["line"], - "type": "eks_version", - "version": version, - "status": EKS_VERSIONS[version]["status"], - "message": warning_msg, - "snippet": finding["snippet"] - }) - + warnings.append( + { + "file": str(file_path), + "line": finding["line"], + "type": "eks_version", + "version": version, + "status": EKS_VERSIONS[version]["status"], + "message": warning_msg, + "snippet": finding["snippet"], + } + ) + return warnings diff --git a/src/shieldcommit/gcp_db_detector.py b/src/shieldcommit/gcp_db_detector.py index 4636129..64ff947 100644 --- a/src/shieldcommit/gcp_db_detector.py +++ b/src/shieldcommit/gcp_db_detector.py @@ -21,7 +21,7 @@ "8.0": {"status": "current", "eol": "2026-04"}, "5.7": {"status": "deprecated", "eol": "2024-10"}, "5.6": {"status": "deprecated", "eol": "2021-02"}, - } + }, }, # PostgreSQL versions "postgres": { @@ -33,7 +33,7 @@ "13": {"status": "extended", "eol": "2025-11"}, "12": {"status": "deprecated", "eol": "2024-11"}, "11": {"status": "deprecated", "eol": "2023-10"}, - } + }, }, # SQL Server versions "sqlserver": { @@ -42,10 +42,11 @@ "2019": {"status": "extended", "eol": "2025-01"}, "2017": {"status": "deprecated", "eol": "2024-10"}, "2016": {"status": "deprecated", "eol": "2024-07"}, - } + }, }, } + def parse_terraform_gcp_cloudsql_versions(content: str) -> List[Dict[str, Any]]: """ Parse Terraform file content for GCP Cloud SQL database versions. @@ -54,17 +55,17 @@ def parse_terraform_gcp_cloudsql_versions(content: str) -> List[Dict[str, Any]]: """ findings = [] lines = content.splitlines() - + # Pattern to match: database_version = "MYSQL_8_0" or "POSTGRES_14" or "SQLSERVER_2019" db_version_pattern = r'database_version\s*=\s*["\']([A-Z_0-9]+)["\']' - + current_resource = None - + for line_no, line in enumerate(lines, 1): # Detect resource type - if 'google_sql_database_instance' in line: - current_resource = 'sql_instance' - + if "google_sql_database_instance" in line: + current_resource = "sql_instance" + # Match database versions if current_resource: db_matches = re.finditer(db_version_pattern, line) @@ -72,19 +73,22 @@ def parse_terraform_gcp_cloudsql_versions(content: str) -> List[Dict[str, Any]]: version_str = match.group(1) # Parse version string like MYSQL_8_0, POSTGRES_14, SQLSERVER_2019 engine, version = parse_gcp_version_string(version_str) - - findings.append({ - "line": line_no, - "engine": engine, - "version": version, - "version_str": version_str, - "snippet": line.strip(), - "match": match.group(0), - "type": "version" - }) - + + findings.append( + { + "line": line_no, + "engine": engine, + "version": version, + "version_str": version_str, + "snippet": line.strip(), + "match": match.group(0), + "type": "version", + } + ) + return findings + def parse_gcp_version_string(version_str: str) -> tuple: """ Parse GCP database version strings. @@ -92,37 +96,40 @@ def parse_gcp_version_string(version_str: str) -> tuple: POSTGRES_14 -> ('postgres', '14') SQLSERVER_2019 -> ('sqlserver', '2019') """ - if version_str.startswith('MYSQL'): + if version_str.startswith("MYSQL"): # MYSQL_5_7 or MYSQL_8_0 - version = version_str.replace('MYSQL_', '').replace('_', '.') - return ('mysql', version) - elif version_str.startswith('POSTGRES'): + version = version_str.replace("MYSQL_", "").replace("_", ".") + return ("mysql", version) + elif version_str.startswith("POSTGRES"): # POSTGRES_11 or POSTGRES_15 - version = version_str.replace('POSTGRES_', '') - return ('postgres', version) - elif version_str.startswith('SQLSERVER'): + version = version_str.replace("POSTGRES_", "") + return ("postgres", version) + elif version_str.startswith("SQLSERVER"): # SQLSERVER_2016 or SQLSERVER_2019 - version = version_str.replace('SQLSERVER_', '') - return ('sqlserver', version) + version = version_str.replace("SQLSERVER_", "") + return ("sqlserver", version) else: # Unknown format - return ('unknown', version_str) + return ("unknown", version_str) + def get_gcp_cloudsql_warning(engine: str, version: str) -> str: """Get warning message for a given GCP Cloud SQL database version.""" if engine not in GCP_CLOUDSQL_ENGINES: return f"⚠️ GCP Cloud SQL engine '{engine}' is not recognized. Check GCP documentation." - + engine_info = GCP_CLOUDSQL_ENGINES[engine] versions = engine_info.get("versions", {}) - + if version not in versions: - return f"⚠️ {engine.upper()} version {version} is unknown. Check GCP Cloud SQL documentation." - + return ( + f"⚠️ {engine.upper()} version {version} is unknown. Check GCP Cloud SQL documentation." + ) + info = versions[version] status = info["status"] eol = info["eol"] - + if status == "deprecated": return f"🚨 GCP Cloud SQL {engine.upper()} {version} is DEPRECATED (EOL: {eol}). Upgrade immediately." elif status == "extended": @@ -130,28 +137,29 @@ def get_gcp_cloudsql_warning(engine: str, version: str) -> str: else: # current return f"✓ GCP Cloud SQL {engine.upper()} {version} is currently supported (EOL: {eol})." + def scan_gcp_cloudsql_versions(file_path: Path) -> List[Dict[str, Any]]: """ Scan a Terraform file for GCP Cloud SQL database version warnings. Returns list of warnings. """ warnings = [] - + # Only scan Terraform files - if file_path.suffix not in {'.tf', '.json'}: + if file_path.suffix not in {".tf", ".json"}: return warnings - + try: content = file_path.read_text(errors="ignore") except Exception: return warnings - + findings = parse_terraform_gcp_cloudsql_versions(content) - + for finding in findings: engine = finding["engine"] version = finding["version"] - + if engine in GCP_CLOUDSQL_ENGINES: versions = GCP_CLOUDSQL_ENGINES[engine].get("versions", {}) if version in versions: @@ -159,15 +167,17 @@ def scan_gcp_cloudsql_versions(file_path: Path) -> List[Dict[str, Any]]: # Only warn on non-current versions if info["status"] != "current": warning_msg = get_gcp_cloudsql_warning(engine, version) - warnings.append({ - "file": str(file_path), - "line": finding["line"], - "type": "gcp_cloudsql_version", - "engine": engine, - "version": version, - "status": info["status"], - "message": warning_msg, - "snippet": finding["snippet"] - }) - + warnings.append( + { + "file": str(file_path), + "line": finding["line"], + "type": "gcp_cloudsql_version", + "engine": engine, + "version": version, + "status": info["status"], + "message": warning_msg, + "snippet": finding["snippet"], + } + ) + return warnings diff --git a/src/shieldcommit/gcp_detector.py b/src/shieldcommit/gcp_detector.py index 4a87a95..a0de000 100644 --- a/src/shieldcommit/gcp_detector.py +++ b/src/shieldcommit/gcp_detector.py @@ -20,6 +20,7 @@ "1.23": {"status": "deprecated", "eol": "2023-12"}, } + def parse_terraform_gcp_versions(content: str) -> List[Dict[str, Any]]: """ Parse Terraform file content for GCP GKE cluster versions. @@ -30,71 +31,78 @@ def parse_terraform_gcp_versions(content: str) -> List[Dict[str, Any]]: """ findings = [] lines = content.splitlines() - + # Pattern 1: Explicit version in resources: min_master_version = "1.27" version_pattern = r'min_master_version\s*=\s*["\']([0-9]+\.[0-9]+)["\']' channel_pattern = r'channel\s*=\s*["\']([A-Z]+)["\']' - + # Pattern 2: Variable default values var_version_pattern = r'default\s*=\s*["\']([0-9]+\.[0-9]+)["\']' - + in_cluster_version_var = False - + for line_no, line in enumerate(lines, 1): # Track if we're in a cluster_version variable block - if 'variable' in line and 'cluster_version' in line: + if "variable" in line and "cluster_version" in line: in_cluster_version_var = True - elif in_cluster_version_var and '}' in line: + elif in_cluster_version_var and "}" in line: in_cluster_version_var = False - + # Check for explicit version in resources version_matches = re.finditer(version_pattern, line) for match in version_matches: version = match.group(1) - findings.append({ - "line": line_no, - "version": version, - "snippet": line.strip(), - "match": match.group(0), - "type": "explicit" - }) - + findings.append( + { + "line": line_no, + "version": version, + "snippet": line.strip(), + "match": match.group(0), + "type": "explicit", + } + ) + # Check for version in variable defaults if in_cluster_version_var: version_matches = re.finditer(var_version_pattern, line) for match in version_matches: version = match.group(1) - findings.append({ - "line": line_no, - "version": version, - "snippet": line.strip(), - "match": match.group(0), - "type": "variable" - }) - + findings.append( + { + "line": line_no, + "version": version, + "snippet": line.strip(), + "match": match.group(0), + "type": "variable", + } + ) + # Check for release channel channel_matches = re.finditer(channel_pattern, line) for match in channel_matches: channel = match.group(1) - findings.append({ - "line": line_no, - "channel": channel, - "snippet": line.strip(), - "match": match.group(0), - "type": "channel" - }) - + findings.append( + { + "line": line_no, + "channel": channel, + "snippet": line.strip(), + "match": match.group(0), + "type": "channel", + } + ) + return findings + def get_version_warning(version: str) -> str: """Get warning message for a given GCP GKE version.""" if version not in GCP_VERSIONS: return f"⚠️ GCP GKE version {version} is unknown. Check Google Cloud documentation." - + info = GCP_VERSIONS[version] status = info["status"] eol = info["eol"] - + if status == "deprecated": return f"🚨 GCP GKE {version} is DEPRECATED (EOL: {eol}). Upgrade immediately." elif status == "extended": @@ -102,59 +110,65 @@ def get_version_warning(version: str) -> str: else: # current return f"✓ GCP GKE {version} is currently supported (EOL: {eol})." + def get_channel_info(channel: str) -> str: """Get info message for GCP release channel.""" channels = { "RAPID": "GCP GKE RAPID channel - receives latest versions quickly.", "REGULAR": "GCP GKE REGULAR channel - recommended for production.", - "STABLE": "GCP GKE STABLE channel - receives latest versions after extended validation." + "STABLE": "GCP GKE STABLE channel - receives latest versions after extended validation.", } return channels.get(channel, f"⚠️ Unknown GCP release channel: {channel}") + def scan_gcp_versions(file_path: Path) -> List[Dict[str, Any]]: """ Scan a Terraform file for GCP GKE version warnings. Returns list of warnings. """ warnings = [] - + # Only scan Terraform files - if file_path.suffix not in {'.tf', '.json'}: + if file_path.suffix not in {".tf", ".json"}: return warnings - + try: content = file_path.read_text(errors="ignore") except Exception: return warnings - + findings = parse_terraform_gcp_versions(content) - + for finding in findings: if finding["type"] == "explicit": version = finding["version"] warning_msg = get_version_warning(version) - + # Only warn on non-current versions if version in GCP_VERSIONS and GCP_VERSIONS[version]["status"] != "current": - warnings.append({ - "file": str(file_path), - "line": finding["line"], - "type": "gcp_version", - "version": version, - "status": GCP_VERSIONS[version]["status"], - "message": warning_msg, - "snippet": finding["snippet"] - }) + warnings.append( + { + "file": str(file_path), + "line": finding["line"], + "type": "gcp_version", + "version": version, + "status": GCP_VERSIONS[version]["status"], + "message": warning_msg, + "snippet": finding["snippet"], + } + ) elif finding["type"] == "channel": channel = finding["channel"] info_msg = get_channel_info(channel) - warnings.append({ - "file": str(file_path), - "line": finding["line"], - "type": "gcp_channel", - "channel": channel, - "message": info_msg, - "snippet": finding["snippet"] - }) - + warnings.append( + { + "file": str(file_path), + "line": finding["line"], + "type": "gcp_channel", + "channel": channel, + "message": info_msg, + "snippet": finding["snippet"], + } + ) + return warnings diff --git a/src/shieldcommit/installer.py b/src/shieldcommit/installer.py index 2659a10..8ef3d93 100644 --- a/src/shieldcommit/installer.py +++ b/src/shieldcommit/installer.py @@ -12,16 +12,18 @@ exit 0 """ -def install_hook(repo_path='.'): - git_hooks = Path(repo_path) / '.git' / 'hooks' + +def install_hook(repo_path="."): + git_hooks = Path(repo_path) / ".git" / "hooks" git_hooks.mkdir(parents=True, exist_ok=True) - hook_file = git_hooks / 'pre-commit' + hook_file = git_hooks / "pre-commit" hook_file.write_text(HOOK_TEMPLATE) hook_file.chmod(0o755) return True -def uninstall_hook(repo_path='.'): - hook_file = Path(repo_path) / '.git' / 'hooks' / 'pre-commit' + +def uninstall_hook(repo_path="."): + hook_file = Path(repo_path) / ".git" / "hooks" / "pre-commit" if hook_file.exists(): hook_file.unlink() return True diff --git a/src/shieldcommit/intelligent_detector.py b/src/shieldcommit/intelligent_detector.py index 4c794bb..834cfee 100644 --- a/src/shieldcommit/intelligent_detector.py +++ b/src/shieldcommit/intelligent_detector.py @@ -14,7 +14,7 @@ class IntelligentDetector: """ Detects secrets using intelligent heuristics instead of patterns. - + Methods: - Entropy analysis: Detects high-entropy strings likely to be secrets - Semantic analysis: Recognizes variable names suggesting secrets @@ -22,68 +22,176 @@ class IntelligentDetector: - Value analysis: Analyzes value characteristics (length, character variety) - Heuristic detection: Applies domain knowledge for common secret formats """ - + # Variable name keywords indicating secrets SECRET_KEYWORDS = { - 'password', 'passwd', 'pwd', 'secret', 'token', 'api_key', 'apikey', - 'api-key', 'key', 'credential', 'credentials', 'auth', 'auth_token', - 'access_token', 'refresh_token', 'api_secret', 'bearer', 'jwt', - 'private_key', 'private-key', 'ssh_key', 'private_token', - 'oauth_token', 'oauth2_token', 'session', 'session_id', 'session-id', - 'aws_secret', 'aws_key', 'aws_access', 'google_key', 'google_secret', - 'github_token', 'gitlab_token', 'bitbucket_token', 'slack_token', - 'stripe_key', 'stripe_secret', 'database_password', 'db_password', - 'db_pass', 'db_user', 'database_user', 'connection_string', - 'connection-string', 'cert', 'certificate', 'pem', 'rsa', 'dsa', - 'cipher', 'encryption_key', 'hmac', 'hash', 'webhook', 'webhook_secret', - 'signing_key', 'signing-key', 'verification_key', 'access_secret', - 'client_secret', 'client-secret', 'client_id', 'app_secret', 'app-secret', - 'master_key', 'master-key', 'admin_password', 'admin-password', - 'root_password', 'root-password', 'user_password', 'user-password' + "password", + "passwd", + "pwd", + "secret", + "token", + "api_key", + "apikey", + "api-key", + "key", + "credential", + "credentials", + "auth", + "auth_token", + "access_token", + "refresh_token", + "api_secret", + "bearer", + "jwt", + "private_key", + "private-key", + "ssh_key", + "private_token", + "oauth_token", + "oauth2_token", + "session", + "session_id", + "session-id", + "aws_secret", + "aws_key", + "aws_access", + "google_key", + "google_secret", + "github_token", + "gitlab_token", + "bitbucket_token", + "slack_token", + "stripe_key", + "stripe_secret", + "database_password", + "db_password", + "db_pass", + "db_user", + "database_user", + "connection_string", + "connection-string", + "cert", + "certificate", + "pem", + "rsa", + "dsa", + "cipher", + "encryption_key", + "hmac", + "hash", + "webhook", + "webhook_secret", + "signing_key", + "signing-key", + "verification_key", + "access_secret", + "client_secret", + "client-secret", + "client_id", + "app_secret", + "app-secret", + "master_key", + "master-key", + "admin_password", + "admin-password", + "root_password", + "root-password", + "user_password", + "user-password", } - + # Context keywords that indicate nearby secrets CONTEXT_KEYWORDS = { - 'password', 'secret', 'token', 'api', 'key', 'auth', 'credential', - 'token:', 'api_key:', 'secret:', 'password:', 'authorization:', - 'bearer:', 'api-key:', 'access-token:', 'auth-token:' + "password", + "secret", + "token", + "api", + "key", + "auth", + "credential", + "token:", + "api_key:", + "secret:", + "password:", + "authorization:", + "bearer:", + "api-key:", + "access-token:", + "auth-token:", } - + # Non-secret indicators (things that shouldn't be flagged) EXCLUDE_KEYWORDS = { - 'arn:', 'arn-', 'policy_arn', 'role_arn', 'resource_arn', - 'account_id', 'account-id', 'account:', 'region:', 'service:', - 'hash:', 'checksum:', 'uuid:', 'id:', 'identifier:', - 'image_id', 'instance_id', 'version:', 'build:', 'release:', - 'name:', 'description:', 'tag:', 'label:', 'namespace:', - 'project_id', 'project-id', 'org_id', 'org-id', - 'user_id', 'user-id', 'admin_id', 'admin-id', - 'correlation_id', 'correlation-id', 'request_id', 'request-id', - 'trace_id', 'trace-id', 'session_id', 'session-id', - 'bucket_id', 'bucket-id', 'cluster_id', 'cluster-id', - 'instance_name', 'resource_name', 'resource_id' + "arn:", + "arn-", + "policy_arn", + "role_arn", + "resource_arn", + "account_id", + "account-id", + "account:", + "region:", + "service:", + "hash:", + "checksum:", + "uuid:", + "id:", + "identifier:", + "image_id", + "instance_id", + "version:", + "build:", + "release:", + "name:", + "description:", + "tag:", + "label:", + "namespace:", + "project_id", + "project-id", + "org_id", + "org-id", + "user_id", + "user-id", + "admin_id", + "admin-id", + "correlation_id", + "correlation-id", + "request_id", + "request-id", + "trace_id", + "trace-id", + "session_id", + "session-id", + "bucket_id", + "bucket-id", + "cluster_id", + "cluster-id", + "instance_name", + "resource_name", + "resource_id", } - + # Common secret structures (formats likely to be secrets) SECRET_STRUCTURES = [ # Base64-like with specific prefixes - (r'^AKIA[0-9A-Z]{16}$', 'AWS Access Key'), # AWS Access Key - (r'^ASIA[0-9A-Z]{16}$', 'AWS Temporary Key'), # AWS Temporary Access Key - (r'^AIza[0-9A-Za-z\-_]{35}$', 'Google API Key'), # Google API Key - (r'^ya29\.[0-9A-Za-z\-_]{1,}$', 'Google OAuth Token'), # Google OAuth - (r'^sk-[A-Za-z0-9]{20,}$', 'OpenAI API Key'), # OpenAI - (r'^pk_live_[0-9a-zA-Z]{24}$', 'Stripe Publishable Key'), # Stripe - (r'^sk_live_[0-9a-zA-Z]{24}$', 'Stripe Secret Key'), # Stripe - (r'^gh[pousr]_[A-Za-z0-9]{30,}$', 'GitHub Token'), # GitHub - (r'^glpat-[0-9a-zA-Z\-]{20,}$', 'GitLab Token'), # GitLab - (r'^bbp_[A-Za-z0-9]{32}$', 'Bitbucket App Password'), # Bitbucket - (r'^xox[baprs]-[0-9A-Za-z]{10,48}$', 'Slack Token'), # Slack - (r'^SG\.[A-Za-z0-9_\-]{22}\.[A-Za-z0-9_\-]{43}$', 'SendGrid API Key'), # SendGrid - (r'^key-[0-9a-zA-Z]{32}$', 'Mailgun API Key'), # Mailgun - (r'^-----BEGIN (RSA|DSA|EC|OPENSSH) PRIVATE KEY-----', 'Private Key'), # SSH Keys - (r'^Bearer\s+[A-Za-z0-9\-._~+/]+=*$', 'Bearer Token'), # Bearer tokens + (r"^AKIA[0-9A-Z]{16}$", "AWS Access Key"), # AWS Access Key + (r"^ASIA[0-9A-Z]{16}$", "AWS Temporary Key"), # AWS Temporary Access Key + (r"^AIza[0-9A-Za-z\-_]{35}$", "Google API Key"), # Google API Key + (r"^ya29\.[0-9A-Za-z\-_]{1,}$", "Google OAuth Token"), # Google OAuth + (r"^sk-[A-Za-z0-9]{20,}$", "OpenAI API Key"), # OpenAI + (r"^pk_live_[0-9a-zA-Z]{24}$", "Stripe Publishable Key"), # Stripe + (r"^sk_live_[0-9a-zA-Z]{24}$", "Stripe Secret Key"), # Stripe + (r"^gh[pousr]_[A-Za-z0-9]{30,}$", "GitHub Token"), # GitHub + (r"^glpat-[0-9a-zA-Z\-]{20,}$", "GitLab Token"), # GitLab + (r"^bbp_[A-Za-z0-9]{32}$", "Bitbucket App Password"), # Bitbucket + (r"^xox[baprs]-[0-9A-Za-z]{10,48}$", "Slack Token"), # Slack + (r"^SG\.[A-Za-z0-9_\-]{22}\.[A-Za-z0-9_\-]{43}$", "SendGrid API Key"), # SendGrid + (r"^key-[0-9a-zA-Z]{32}$", "Mailgun API Key"), # Mailgun + (r"^-----BEGIN (RSA|DSA|EC|OPENSSH) PRIVATE KEY-----", "Private Key"), # SSH Keys + (r"^Bearer\s+[A-Za-z0-9\-._~+/]+=*$", "Bearer Token"), # Bearer tokens ] - + @staticmethod def calculate_entropy(s: str) -> float: """ @@ -93,18 +201,18 @@ def calculate_entropy(s: str) -> float: """ if not s or len(s) < 1: return 0.0 - + # Count unique characters unique_chars = len(set(s)) - + # Calculate probabilities and entropy entropy = 0.0 for char in set(s): probability = s.count(char) / len(s) entropy -= probability * math.log2(probability) - + return entropy - + @staticmethod def is_high_entropy(value: str, min_entropy: float = 3.5, min_length: int = 12) -> bool: """ @@ -116,21 +224,21 @@ def is_high_entropy(value: str, min_entropy: float = 3.5, min_length: int = 12) """ if len(value) < min_length: return False - + entropy = IntelligentDetector.calculate_entropy(value) if entropy < min_entropy: return False - + # Check character variety (should have different types) - has_alpha = bool(re.search(r'[a-zA-Z]', value)) - has_digit = bool(re.search(r'\d', value)) - has_special = bool(re.search(r'[^a-zA-Z0-9]', value)) - + has_alpha = bool(re.search(r"[a-zA-Z]", value)) + has_digit = bool(re.search(r"\d", value)) + has_special = bool(re.search(r"[^a-zA-Z0-9]", value)) + # Need at least 2 of 3 character types for high entropy to be meaningful variety = sum([has_alpha, has_digit, has_special]) - + return variety >= 2 - + @staticmethod def is_likely_secret_format(value: str) -> Tuple[bool, str]: """ @@ -140,9 +248,9 @@ def is_likely_secret_format(value: str) -> Tuple[bool, str]: for pattern, name in IntelligentDetector.SECRET_STRUCTURES: if re.match(pattern, value): return True, name - + return False, "" - + @staticmethod def extract_secret_indicator(line: str) -> Tuple[str, float]: """ @@ -153,16 +261,16 @@ def extract_secret_indicator(line: str) -> Tuple[str, float]: match = re.search(r'([a-zA-Z_][a-zA-Z0-9_\-]*)\s*[:=]\s*["\']?([^"\';\s]+)', line) if not match: return "", 0.0 - + var_name = match.group(1).lower() - + # Check if variable name suggests a secret for keyword in IntelligentDetector.SECRET_KEYWORDS: if keyword in var_name: return var_name, 0.3 # Boost confidence - + return var_name, 0.0 - + @staticmethod def has_secret_context(line: str) -> bool: """ @@ -170,7 +278,7 @@ def has_secret_context(line: str) -> bool: """ line_lower = line.lower() return any(keyword in line_lower for keyword in IntelligentDetector.CONTEXT_KEYWORDS) - + @staticmethod def is_excluded(value: str, line: str) -> bool: """ @@ -179,228 +287,233 @@ def is_excluded(value: str, line: str) -> bool: """ value_lower = value.lower() line_lower = line.lower() - + # ============ INFRASTRUCTURE AS CODE PATTERNS ============ - + # Terraform/CloudFormation interpolations - if '${' in value or '}' in value or '{%' in value: + if "${" in value or "}" in value or "{%" in value: return True - + # Variable references (Terraform, CloudFormation, etc.) - if 'var.' in value_lower or 'local.' in value_lower or 'data.' in value_lower: + if "var." in value_lower or "local." in value_lower or "data." in value_lower: return True - if 'data:' in value_lower or 'module.' in value_lower: + if "data:" in value_lower or "module." in value_lower: return True - + # Terraform resource references (resource_type.resource_name.attribute) - if re.match(r'^[a-z_]+\.[a-z_]+(\.[a-z_]+)*$', value_lower): + if re.match(r"^[a-z_]+\.[a-z_]+(\.[a-z_]+)*$", value_lower): # This matches patterns like: aws_vpc.main.id, aws_db_instance.postgres.endpoint - if any(keyword in value_lower for keyword in ['aws_', 'google_', 'azurerm_', 'kubernetes_']): + if any( + keyword in value_lower for keyword in ["aws_", "google_", "azurerm_", "kubernetes_"] + ): return True - + # Kubernetes patterns - if 'kind:' in value_lower or 'apiversion:' in value_lower: + if "kind:" in value_lower or "apiversion:" in value_lower: return True - if value_lower.startswith('k8s-') or value_lower.startswith('kube-'): + if value_lower.startswith("k8s-") or value_lower.startswith("kube-"): return True - + # ============ AWS IDENTIFIERS ============ - - if value_lower.startswith('arn:') or 'arn:' in line_lower: + + if value_lower.startswith("arn:") or "arn:" in line_lower: return True - if value_lower.startswith('i-') and len(value) == 19: # EC2 instance ID + if value_lower.startswith("i-") and len(value) == 19: # EC2 instance ID return True - if value_lower.startswith('ami-') or value_lower.startswith('ami_'): # AMI ID + if value_lower.startswith("ami-") or value_lower.startswith("ami_"): # AMI ID return True - if value_lower.startswith('sg-'): # Security group + if value_lower.startswith("sg-"): # Security group return True - if value_lower.startswith('subnet-'): # Subnet + if value_lower.startswith("subnet-"): # Subnet return True - if value_lower.startswith('vpc-'): # VPC + if value_lower.startswith("vpc-"): # VPC return True - if value_lower.startswith('vol-'): # Volume + if value_lower.startswith("vol-"): # Volume return True - if value_lower.startswith('snap-'): # Snapshot + if value_lower.startswith("snap-"): # Snapshot return True - if value_lower.startswith('rtb-'): # Route table + if value_lower.startswith("rtb-"): # Route table return True - if value_lower.startswith('acl-'): # Network ACL + if value_lower.startswith("acl-"): # Network ACL return True - if value_lower.startswith('eni-'): # Network interface + if value_lower.startswith("eni-"): # Network interface return True - if value_lower.startswith('nat-'): # NAT gateway + if value_lower.startswith("nat-"): # NAT gateway return True - if value_lower.startswith('igw-'): # Internet gateway + if value_lower.startswith("igw-"): # Internet gateway return True - if value_lower.startswith('eip-'): # Elastic IP + if value_lower.startswith("eip-"): # Elastic IP return True - if value_lower.startswith('dopt-'): # DHCP options set + if value_lower.startswith("dopt-"): # DHCP options set return True - + # ============ GCP IDENTIFIERS ============ - - if value.startswith('projects/'): # GCP resource + + if value.startswith("projects/"): # GCP resource return True - if value_lower.startswith('gke-'): # GKE cluster + if value_lower.startswith("gke-"): # GKE cluster return True - if re.match(r'^\d{10,}$', value): # GCP project number + if re.match(r"^\d{10,}$", value): # GCP project number return True - + # ============ AZURE IDENTIFIERS ============ - - if value_lower.startswith('/subscriptions/'): # Azure resource path + + if value_lower.startswith("/subscriptions/"): # Azure resource path return True - if value_lower.startswith('guid-') or value_lower.startswith('id-'): + if value_lower.startswith("guid-") or value_lower.startswith("id-"): return True - + # ============ STANDARD IDENTIFIERS ============ - + # UUID - if re.match(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', value_lower): + if re.match(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", value_lower): return True - + # Hashes (MD5, SHA1, SHA256, SHA512) - if re.match(r'^[a-f0-9]{32}$|^[a-f0-9]{40}$|^[a-f0-9]{64}$|^[a-f0-9]{128}$', value_lower): + if re.match(r"^[a-f0-9]{32}$|^[a-f0-9]{40}$|^[a-f0-9]{64}$|^[a-f0-9]{128}$", value_lower): return True - + # Git hashes - if re.match(r'^[0-9a-f]{7,40}$', value_lower) and 'git' in line_lower: + if re.match(r"^[0-9a-f]{7,40}$", value_lower) and "git" in line_lower: return True - + # Docker SHA - if value_lower.startswith('sha256:'): + if value_lower.startswith("sha256:"): return True - if value_lower.startswith('sha512:'): + if value_lower.startswith("sha512:"): return True - + # ============ NETWORKING & DNS ============ - + # URLs and endpoints - if value_lower.startswith('http://') or value_lower.startswith('https://'): + if value_lower.startswith("http://") or value_lower.startswith("https://"): return True - if value_lower.startswith('ws://') or value_lower.startswith('wss://'): + if value_lower.startswith("ws://") or value_lower.startswith("wss://"): return True - if value_lower.startswith('grpc://'): + if value_lower.startswith("grpc://"): return True - if value_lower.startswith('file://') or value_lower.startswith('ftp://'): + if value_lower.startswith("file://") or value_lower.startswith("ftp://"): return True - + # Domain names and FQDNs - if re.match(r'^([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}$', value_lower): + if re.match(r"^([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}$", value_lower): return True - + # Email addresses - if '@' in value and '.' in value: - if re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', value): + if "@" in value and "." in value: + if re.match(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$", value): return True - + # IP addresses - if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', value): # IPv4 + if re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", value): # IPv4 return True - if ':' in value and re.match(r'^[0-9a-f:]+$', value_lower): # IPv6 - if value_lower.count(':') >= 2: + if ":" in value and re.match(r"^[0-9a-f:]+$", value_lower): # IPv6 + if value_lower.count(":") >= 2: return True - + # Ports in URLs/addresses - if re.match(r'^\d{1,5}$', value) and int(value) <= 65535: + if re.match(r"^\d{1,5}$", value) and int(value) <= 65535: return True - + # ============ CONTAINER & IMAGE IDENTIFIERS ============ - + # Docker image names (repo/name:tag) - if re.match(r'^[a-z0-9\-._/]+:[a-z0-9\-._]+$', value_lower): + if re.match(r"^[a-z0-9\-._/]+:[a-z0-9\-._]+$", value_lower): return True - if re.match(r'^[a-z0-9\-._/]+@sha256:[a-f0-9]{64}$', value_lower): + if re.match(r"^[a-z0-9\-._/]+@sha256:[a-f0-9]{64}$", value_lower): return True - + # Container registry hosts - if 'docker.io' in value_lower or 'gcr.io' in value_lower: + if "docker.io" in value_lower or "gcr.io" in value_lower: return True - if 'azurecr.io' in value_lower or 'ecr.aws' in value_lower: + if "azurecr.io" in value_lower or "ecr.aws" in value_lower: return True - + # ============ PACKAGE & MODULE REFERENCES ============ - + # Go imports - if 'github.com/' in value_lower or 'gitlab.com/' in value_lower: + if "github.com/" in value_lower or "gitlab.com/" in value_lower: return True - + # Python packages - if 'pypi.org' in value_lower or 'python.org' in value_lower: + if "pypi.org" in value_lower or "python.org" in value_lower: return True - + # Java packages - if re.match(r'^[a-z]+(\.[a-z0-9]+)+$', value_lower): - if 'java' in line_lower or 'package' in line_lower: + if re.match(r"^[a-z]+(\.[a-z0-9]+)+$", value_lower): + if "java" in line_lower or "package" in line_lower: return True - + # ============ FILE PATHS & DIRECTORIES ============ - + # File paths - if '/' in value or '\\' in value: - if any(ext in value.lower() for ext in ['.tf', '.yaml', '.yml', '.json', '.xml', '.py', '.js', '.go']): + if "/" in value or "\\" in value: + if any( + ext in value.lower() + for ext in [".tf", ".yaml", ".yml", ".json", ".xml", ".py", ".js", ".go"] + ): return True - + # Kubernetes paths - if value.startswith('/var/run/secrets') or value.startswith('/etc/'): + if value.startswith("/var/run/secrets") or value.startswith("/etc/"): return True - + # ============ VERSION & BUILD INFO ============ - + # Semantic versions (1.2.3) - if re.match(r'^\d+\.\d+(\.\d+)?(\-[a-z0-9]+)?$', value_lower): + if re.match(r"^\d+\.\d+(\.\d+)?(\-[a-z0-9]+)?$", value_lower): return True - + # Build numbers and dates - if re.match(r'^\d{4}-\d{2}-\d{2}', value): # Dates + if re.match(r"^\d{4}-\d{2}-\d{2}", value): # Dates return True - if re.match(r'^\d{10,}$', value): # Unix timestamps + if re.match(r"^\d{10,}$", value): # Unix timestamps return True - + # Commit SHAs in git refs - if 'commit' in line_lower and re.match(r'^[0-9a-f]{7,}$', value_lower): + if "commit" in line_lower and re.match(r"^[0-9a-f]{7,}$", value_lower): return True - + # ============ NAMING PATTERNS ============ - + # Hyphenated naming (word-word-word, word-word-123) - only if ALL lowercase and short - if re.match(r'^[a-z0-9]+(-[a-z0-9]+)*(-\d+)?$', value_lower) and value == value_lower: + if re.match(r"^[a-z0-9]+(-[a-z0-9]+)*(-\d+)?$", value_lower) and value == value_lower: # Only exclude if it looks like a resource name, not a secret if len(value) < 24: # Resource names are typically shorter return True - + # Camel case naming - if re.match(r'^[A-Z][a-z]+([A-Z][a-z]+)*$', value): + if re.match(r"^[A-Z][a-z]+([A-Z][a-z]+)*$", value): return True - + # ============ CONTEXT-BASED EXCLUSIONS ============ - + # Exclude if has exclude keywords in context for keyword in IntelligentDetector.EXCLUDE_KEYWORDS: if keyword in line_lower: return True - + # Exclude if value is mostly digits (IDs, counts) if len(value) > 5 and sum(c.isdigit() for c in value) / len(value) > 0.7: return True - + # ============ RESOURCE NAMING PATTERNS ============ - + # Common resource name patterns: env-region-number - if re.match(r'^[a-z]+-[a-z]+-\d+$', value_lower): + if re.match(r"^[a-z]+-[a-z]+-\d+$", value_lower): return True - + # DNS-like patterns: subdomain.service.namespace - if re.match(r'^[a-z0-9]+\.[a-z0-9]+(\.[a-z0-9]+)*$', value_lower): + if re.match(r"^[a-z0-9]+\.[a-z0-9]+(\.[a-z0-9]+)*$", value_lower): return True - + return False - + @staticmethod def calculate_confidence(value: str, line: str = "") -> float: """ Calculate confidence score (0.0 to 1.0) that value is a secret. - + Factors: - Entropy analysis: High entropy increases confidence - Format matching: Known secret formats increase confidence @@ -411,173 +524,182 @@ def calculate_confidence(value: str, line: str = "") -> float: # Exclusion check: If excluded, confidence is 0 if IntelligentDetector.is_excluded(value, line): return 0.0 - + confidence = 0.0 - + # 1. Format matching (high confidence) is_known_format, format_name = IntelligentDetector.is_likely_secret_format(value) if is_known_format: return 0.95 # Known formats are very confident - + # 2. Entropy analysis (moderate to high confidence) if IntelligentDetector.is_high_entropy(value): entropy = IntelligentDetector.calculate_entropy(value) # Map entropy 3.5-5.7 to confidence 0.4-0.8 confidence += min(0.8, (entropy - 3.5) / 2.2 * 0.8) - + # 3. Semantic analysis from variable name (moderate confidence) var_indicator, indicator_boost = IntelligentDetector.extract_secret_indicator(line) if indicator_boost > 0: confidence += indicator_boost - + # 4. Context analysis (low to moderate confidence) if line and IntelligentDetector.has_secret_context(line): confidence += 0.15 - + # 5. Length analysis (longer strings more likely to be secrets) if len(value) >= 32: confidence += 0.1 elif len(value) >= 20: confidence += 0.05 - + # 6. Value analysis (special characters, mix of types) - has_alpha = bool(re.search(r'[a-zA-Z]', value)) - has_digit = bool(re.search(r'\d', value)) - has_special = bool(re.search(r'[^a-zA-Z0-9]', value)) + has_alpha = bool(re.search(r"[a-zA-Z]", value)) + has_digit = bool(re.search(r"\d", value)) + has_special = bool(re.search(r"[^a-zA-Z0-9]", value)) variety = sum([has_alpha, has_digit, has_special]) - + if variety == 3: confidence += 0.1 - + # Cap confidence at 1.0 return min(1.0, confidence) - + @staticmethod def detect_in_line(line: str, min_confidence: float = 0.5, context: str = "") -> List[Dict]: """ Detect potential secrets in a single line. Returns list of findings with confidence scores. - + Skips common production patterns: Terraform interpolations, URLs, paths, etc. - + Args: line: The line to scan min_confidence: Minimum confidence threshold context: Previous lines for variable name context (e.g., "variable db_password") """ findings = [] - + # Skip lines with Terraform/CloudFormation interpolations (naming patterns) - if '${' in line or '}' in line or '{%' in line: + if "${" in line or "}" in line or "{%" in line: return findings - + # Skip URLs, paths, and references - if any(proto in line for proto in ['http://', 'https://', 'ws://', 'wss://', 'file://', 'ftp://']): + if any( + proto in line + for proto in ["http://", "https://", "ws://", "wss://", "file://", "ftp://"] + ): return findings - + # Skip Kubernetes resources and manifests - if 'kind:' in line.lower() or 'apiversion:' in line.lower(): + if "kind:" in line.lower() or "apiversion:" in line.lower(): return findings - + # Skip common comment lines with examples - if '#' in line or '//' in line: + if "#" in line or "//" in line: return findings - + # Extract potential values from assignments # Patterns: var = "value", var = value, var: "value", var: value patterns = [ r'([a-zA-Z_][a-zA-Z0-9_\-]*)\s*[:=]\s*["\']([^"\']+)["\']', # Quoted - r'([a-zA-Z_][a-zA-Z0-9_\-]*)\s*[:=]\s*([A-Za-z0-9\-_.+/]+)', # Unquoted + r"([a-zA-Z_][a-zA-Z0-9_\-]*)\s*[:=]\s*([A-Za-z0-9\-_.+/]+)", # Unquoted ] - + for pattern in patterns: for match in re.finditer(pattern, line): var_name = match.group(1) value = match.group(2) - + # Skip very short values if len(value) < 8: continue - + # Skip common non-secrets - if value in ['true', 'false', 'null', 'none', 'yes', 'no']: + if value in ["true", "false", "null", "none", "yes", "no"]: continue - + # Skip if value contains variable references or interpolations - if 'var.' in value or 'local.' in value or 'data.' in value: + if "var." in value or "local." in value or "data." in value: continue - + # Use context-aware confidence calculation full_line_context = context + " " + line if context else line confidence = IntelligentDetector.calculate_confidence(value, full_line_context) - + if confidence >= min_confidence: - findings.append({ - 'value': value, - 'variable': var_name, - 'confidence': confidence, - 'detection_method': IntelligentDetector._get_detection_method(value, full_line_context) - }) - + findings.append( + { + "value": value, + "variable": var_name, + "confidence": confidence, + "detection_method": IntelligentDetector._get_detection_method( + value, full_line_context + ), + } + ) + return findings - + @staticmethod def _get_detection_method(value: str, line: str) -> str: """Determine which detection method identified the secret.""" is_known, fmt = IntelligentDetector.is_likely_secret_format(value) if is_known: return f"Format: {fmt}" - + if IntelligentDetector.is_high_entropy(value): return "High Entropy" - + var_name, boost = IntelligentDetector.extract_secret_indicator(line) if boost > 0: return f"Semantic: {var_name}" - + if IntelligentDetector.has_secret_context(line): return "Context Analysis" - + return "Heuristic Analysis" def detect_secrets(text: str, min_confidence: float = 0.5) -> List[Dict]: """ Detect secrets in text using intelligent analysis. - + Args: text: Content to scan min_confidence: Minimum confidence threshold (0.0-1.0) - + Returns: List of detected secrets with confidence scores """ findings = [] lines = text.splitlines() - + for line_no, line in enumerate(lines, 1): # Skip empty lines and comments - if not line.strip() or line.strip().startswith('#'): + if not line.strip() or line.strip().startswith("#"): continue - + # Build context from previous lines (look back 3 lines for variable declarations) context = line for i in range(max(0, line_no - 4), line_no - 1): if i < len(lines): context = lines[i] + " " + context - + line_findings = IntelligentDetector.detect_in_line(line, min_confidence, context) - + for finding in line_findings: - findings.append({ - 'file': '', # Will be set by scanner - 'line': line_no, - 'snippet': line[:200], - 'matched_value': finding['value'], - 'variable': finding['variable'], - 'confidence': finding['confidence'], - 'detection_method': finding['detection_method'], - 'pattern': f"Intelligent Detection: {finding['detection_method']}" - }) - + findings.append( + { + "file": "", # Will be set by scanner + "line": line_no, + "snippet": line[:200], + "matched_value": finding["value"], + "variable": finding["variable"], + "confidence": finding["confidence"], + "detection_method": finding["detection_method"], + "pattern": f"Intelligent Detection: {finding['detection_method']}", + } + ) + return findings diff --git a/src/shieldcommit/rds_detector.py b/src/shieldcommit/rds_detector.py index 413b9ab..f96c835 100644 --- a/src/shieldcommit/rds_detector.py +++ b/src/shieldcommit/rds_detector.py @@ -34,6 +34,7 @@ }, } + def parse_terraform_rds_versions(content: str) -> List[Dict[str, Any]]: """ Parse Terraform file content for RDS engine versions. @@ -44,70 +45,77 @@ def parse_terraform_rds_versions(content: str) -> List[Dict[str, Any]]: """ findings = [] lines = content.splitlines() - + # Pattern to match engine and engine_version in resources engine_pattern = r'engine\s*=\s*["\']([a-z]+)["\']' version_pattern = r'engine_version\s*=\s*["\']([0-9.]+)["\']' - + # Pattern for variable defaults var_version_pattern = r'default\s*=\s*["\']([0-9.]+)["\']' - + current_engine = None in_db_version_var = False - + for line_no, line in enumerate(lines, 1): # Track if we're in a db version variable block - if 'variable' in line and ('db_engine_version' in line or 'engine_version' in line or 'db_version' in line): + if "variable" in line and ( + "db_engine_version" in line or "engine_version" in line or "db_version" in line + ): in_db_version_var = True - elif in_db_version_var and '}' in line: + elif in_db_version_var and "}" in line: in_db_version_var = False - + # Check for engine in resources engine_match = re.search(engine_pattern, line) if engine_match: current_engine = engine_match.group(1) - + # Check for version in resources version_match = re.search(version_pattern, line) if version_match and current_engine: version = version_match.group(1) - findings.append({ - "line": line_no, - "engine": current_engine, - "version": version, - "snippet": line.strip(), - "match": version_match.group(0) - }) - + findings.append( + { + "line": line_no, + "engine": current_engine, + "version": version, + "snippet": line.strip(), + "match": version_match.group(0), + } + ) + # Check for version in variable defaults if in_db_version_var: version_match = re.search(var_version_pattern, line) if version_match: # Use postgres as default engine if not found - engine = current_engine or 'postgres' + engine = current_engine or "postgres" version = version_match.group(1) - findings.append({ - "line": line_no, - "engine": engine, - "version": version, - "snippet": line.strip(), - "match": version_match.group(0) - }) - + findings.append( + { + "line": line_no, + "engine": engine, + "version": version, + "snippet": line.strip(), + "match": version_match.group(0), + } + ) + return findings + def get_rds_warning(engine: str, version: str) -> str: """Get warning message for a given RDS engine version.""" if engine not in RDS_ENGINES: return f"⚠️ RDS engine '{engine}' is not recognized. Check AWS documentation." - + if version not in RDS_ENGINES[engine]: return f"⚠️ {engine.upper()} version {version} is unknown. Check AWS documentation." - + info = RDS_ENGINES[engine][version] status = info["status"] eol = info["eol"] - + if status == "deprecated": return f"🚨 {engine.upper()} {version} is DEPRECATED (EOL: {eol}). Upgrade immediately." elif status == "extended": @@ -115,42 +123,45 @@ def get_rds_warning(engine: str, version: str) -> str: else: # current return f"✓ {engine.upper()} {version} is currently supported (EOL: {eol})." + def scan_rds_versions(file_path: Path) -> List[Dict[str, Any]]: """ Scan a Terraform file for RDS version warnings. Returns list of warnings. """ warnings = [] - + # Only scan Terraform files - if file_path.suffix not in {'.tf', '.json'}: + if file_path.suffix not in {".tf", ".json"}: return warnings - + try: content = file_path.read_text(errors="ignore") except Exception: return warnings - + findings = parse_terraform_rds_versions(content) - + for finding in findings: engine = finding["engine"] version = finding["version"] - + if engine in RDS_ENGINES and version in RDS_ENGINES[engine]: info = RDS_ENGINES[engine][version] # Only warn on non-current versions if info["status"] != "current": warning_msg = get_rds_warning(engine, version) - warnings.append({ - "file": str(file_path), - "line": finding["line"], - "type": "rds_version", - "engine": engine, - "version": version, - "status": info["status"], - "message": warning_msg, - "snippet": finding["snippet"] - }) - + warnings.append( + { + "file": str(file_path), + "line": finding["line"], + "type": "rds_version", + "engine": engine, + "version": version, + "status": info["status"], + "message": warning_msg, + "snippet": finding["snippet"], + } + ) + return warnings diff --git a/src/shieldcommit/scanner.py b/src/shieldcommit/scanner.py index 1a114ff..e4136e8 100644 --- a/src/shieldcommit/scanner.py +++ b/src/shieldcommit/scanner.py @@ -7,6 +7,7 @@ from .azure_db_detector import scan_azure_db_versions from .gcp_db_detector import scan_gcp_cloudsql_versions + def scan_file(path: Path, min_confidence: float = 0.5): """ Scan a file for secrets using intelligent detection. @@ -20,9 +21,9 @@ def scan_file(path: Path, min_confidence: float = 0.5): # Use intelligent detection instead of patterns detected = detect_secrets(text, min_confidence=min_confidence) - + for finding in detected: - finding['file'] = str(path) + finding["file"] = str(path) findings.append(finding) return findings @@ -36,7 +37,7 @@ def scan_files(paths): """ findings = [] warnings = [] - + for p in paths: p = Path(p) if p.is_file(): @@ -49,9 +50,5 @@ def scan_files(paths): warnings.extend(scan_rds_versions(p)) warnings.extend(scan_azure_db_versions(p)) warnings.extend(scan_gcp_cloudsql_versions(p)) - - return { - "findings": findings, - "warnings": warnings - } + return {"findings": findings, "warnings": warnings} diff --git a/src/shieldcommit/utils.py b/src/shieldcommit/utils.py index dbad3b6..ca74ce8 100644 --- a/src/shieldcommit/utils.py +++ b/src/shieldcommit/utils.py @@ -1,5 +1,6 @@ import math + def shannon_entropy(s: str) -> float: """Compute Shannon entropy for a string.""" if not s: @@ -7,5 +8,6 @@ def shannon_entropy(s: str) -> float: probs = [s.count(c) / len(s) for c in set(s)] return -sum(p * math.log2(p) for p in probs) + def is_high_entropy(s: str, threshold=4.0) -> bool: return shannon_entropy(s) >= threshold diff --git a/tests/test_aks_detector.py b/tests/test_aks_detector.py index 31b2686..0b68c22 100644 --- a/tests/test_aks_detector.py +++ b/tests/test_aks_detector.py @@ -11,67 +11,76 @@ class TestAKSVersionDetection: """Test Azure AKS version detection and deprecation warnings""" - + def test_detects_deprecated_version(self): """Should detect deprecated Kubernetes versions in AKS""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_kubernetes_cluster" "main" { kubernetes_version = "1.23" } -''') - +""" + ) + findings = scan_aks_versions(file_path) # Version 1.23 should be flagged assert isinstance(findings, list) - + def test_detects_extended_support_version(self): """Should detect versions in extended support period""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_kubernetes_cluster" "aks" { kubernetes_version = "1.25" } -''') - +""" + ) + findings = scan_aks_versions(file_path) assert isinstance(findings, list) - + def test_no_warning_for_current_version(self): """Should NOT warn for current/recent versions""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_kubernetes_cluster" "main" { kubernetes_version = "1.29" } -''') - +""" + ) + findings = scan_aks_versions(file_path) assert len(findings) == 0 - + def test_finds_version_in_variable_block(self): """Should find versions declared in variable blocks""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "variables.tf" - file_path.write_text(''' + file_path.write_text( + """ variable "aks_version" { description = "AKS cluster version" default = "1.24" } -''') - +""" + ) + findings = scan_aks_versions(file_path) # Version 1.24 is deprecated assert isinstance(findings, list) - + def test_multiple_clusters_in_file(self): """Should detect deprecated versions in multiple clusters""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_kubernetes_cluster" "cluster1" { kubernetes_version = "1.23" } @@ -79,22 +88,25 @@ def test_multiple_clusters_in_file(self): resource "azurerm_kubernetes_cluster" "cluster2" { kubernetes_version = "1.24" } -''') - +""" + ) + findings = scan_aks_versions(file_path) assert isinstance(findings, list) - + def test_aks_specific_fields(self): """Should detect AKS-specific configuration patterns""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "aks.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_kubernetes_cluster" "main" { name = "my-aks" kubernetes_version = "1.22" node_resource_group = "MC_rg_aks_eastus" } -''') - +""" + ) + findings = scan_aks_versions(file_path) assert isinstance(findings, list) diff --git a/tests/test_azure_db_detector.py b/tests/test_azure_db_detector.py index d9b4b43..7467fe9 100644 --- a/tests/test_azure_db_detector.py +++ b/tests/test_azure_db_detector.py @@ -11,94 +11,107 @@ class TestAzureDBVersionDetection: """Test Azure Database version detection and deprecation warnings""" - + def test_detects_deprecated_sql_server_version(self): """Should detect deprecated SQL Server versions on Azure""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_mssql_server" "primary" { version = "12.0" } -''') - +""" + ) + findings = scan_azure_db_versions(file_path) # SQL Server 12.0 should be flagged assert isinstance(findings, list) - + def test_detects_deprecated_mysql_version(self): """Should detect deprecated MySQL versions on Azure""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_mysql_server" "primary" { version = "5.7" } -''') - +""" + ) + findings = scan_azure_db_versions(file_path) assert isinstance(findings, list) - + def test_detects_deprecated_postgres_version(self): """Should detect deprecated PostgreSQL versions on Azure""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_postgresql_server" "primary" { version = "10" } -''') - +""" + ) + findings = scan_azure_db_versions(file_path) assert isinstance(findings, list) - + def test_no_warning_for_current_version(self): """Should report current versions as extended support""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_postgresql_server" "primary" { version = "14" } -''') - +""" + ) + findings = scan_azure_db_versions(file_path) # PostgreSQL 14 is on Extended Support, detector still reports it assert len(findings) >= 1 assert "Extended Support" in findings[0]["message"] - + def test_finds_version_in_flexible_server(self): """Should find versions in Azure flexible server""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_postgresql_flexible_server" "primary" { version = "11" } -''') - +""" + ) + findings = scan_azure_db_versions(file_path) assert isinstance(findings, list) - + def test_finds_version_in_variable_block(self): """Should find database versions in variable blocks""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "variables.tf" - file_path.write_text(''' + file_path.write_text( + """ variable "db_version" { description = "Azure database version" default = "11" } -''') - +""" + ) + findings = scan_azure_db_versions(file_path) assert isinstance(findings, list) - + def test_multiple_databases_in_file(self): """Should detect deprecated versions in multiple databases""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_postgresql_server" "postgres" { version = "10" } @@ -110,16 +123,18 @@ def test_multiple_databases_in_file(self): resource "azurerm_mssql_server" "sql" { version = "12.0" } -''') - +""" + ) + findings = scan_azure_db_versions(file_path) assert isinstance(findings, list) - + def test_azure_specific_fields(self): """Should detect Azure-specific database configuration""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "azure_db.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "azurerm_postgresql_server" "primary" { name = "my-postgres" location = "East US" @@ -128,8 +143,9 @@ def test_azure_specific_fields(self): backup_retention_days = 7 geo_redundant_backup_enabled = true } -''') - +""" + ) + findings = scan_azure_db_versions(file_path) # PostgreSQL 9.5 is deprecated assert isinstance(findings, list) diff --git a/tests/test_eks_detector.py b/tests/test_eks_detector.py index 9b549a3..eda6335 100644 --- a/tests/test_eks_detector.py +++ b/tests/test_eks_detector.py @@ -11,95 +11,106 @@ class TestEKSVersionDetection: """Test AWS EKS version detection and deprecation warnings""" - + def test_detects_deprecated_version(self): """Should detect deprecated Kubernetes versions in EKS""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_eks_cluster" "main" { kubernetes_version = "1.24" } -''') - +""" + ) + findings = scan_eks_versions(file_path) assert len(findings) > 0 # Kubernetes 1.24 should be flagged - + def test_detects_extended_support_version(self): """Should detect versions in extended support period""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_eks_cluster" "main" { kubernetes_version = "1.27" } -''') - +""" + ) + findings = scan_eks_versions(file_path) assert len(findings) > 0 # Kubernetes 1.27 should be flagged - + def test_no_warning_for_current_version(self): """Should NOT warn for current/recent versions""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_eks_cluster" "main" { kubernetes_version = "1.30" } -''') - +""" + ) + findings = scan_eks_versions(file_path) assert len(findings) == 0 - + def test_finds_version_in_variable_block(self): """Should find versions declared in variable blocks""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "variables.tf" - file_path.write_text(''' + file_path.write_text( + """ variable "cluster_version" { description = "EKS cluster version" default = "1.25" } -''') - +""" + ) + findings = scan_eks_versions(file_path) assert len(findings) > 0 - + def test_includes_line_numbers(self): """Should include line numbers in findings""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ # Line 1 # Line 2 resource "aws_eks_cluster" "main" { kubernetes_version = "1.24" } -''') - +""" + ) + findings = scan_eks_versions(file_path) assert len(findings) > 0 - assert 'line' in findings[0] - assert findings[0]['line'] == 5 - + assert "line" in findings[0] + assert findings[0]["line"] == 5 + def test_includes_file_path(self): """Should include file path in findings""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" file_path.write_text('kubernetes_version = "1.23"') - + findings = scan_eks_versions(file_path) assert len(findings) > 0 - assert 'file' in findings[0] - assert 'main.tf' in findings[0]['file'] - + assert "file" in findings[0] + assert "main.tf" in findings[0]["file"] + def test_multiple_versions_in_file(self): """Should detect all deprecated versions in file""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_eks_cluster" "cluster1" { kubernetes_version = "1.23" } @@ -107,7 +118,8 @@ def test_multiple_versions_in_file(self): resource "aws_eks_cluster" "cluster2" { kubernetes_version = "1.24" } -''') - +""" + ) + findings = scan_eks_versions(file_path) assert len(findings) >= 2 diff --git a/tests/test_gcp_db_detector.py b/tests/test_gcp_db_detector.py index 167063b..dcf19da 100644 --- a/tests/test_gcp_db_detector.py +++ b/tests/test_gcp_db_detector.py @@ -11,79 +11,90 @@ class TestGCPCloudSQLVersionDetection: """Test Google Cloud SQL version detection and deprecation warnings""" - + def test_detects_deprecated_postgres_version(self): """Should detect deprecated PostgreSQL versions on Cloud SQL""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_sql_database_instance" "primary" { database_version = "POSTGRES_11" } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) # PostgreSQL 11 is deprecated on Cloud SQL assert isinstance(findings, list) - + def test_detects_deprecated_mysql_version(self): """Should detect deprecated MySQL versions on Cloud SQL""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_sql_database_instance" "primary" { database_version = "MYSQL_5_7" } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) assert isinstance(findings, list) - + def test_detects_deprecated_sql_server_version(self): """Should detect deprecated SQL Server versions on Cloud SQL""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_sql_database_instance" "primary" { database_version = "SQLSERVER_2016" } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) assert isinstance(findings, list) - + def test_no_warning_for_current_version(self): """Should NOT warn for current Cloud SQL versions""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_sql_database_instance" "primary" { database_version = "POSTGRES_15" } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) assert len(findings) == 0 - + def test_finds_version_in_variable_block(self): """Should find database versions in variable blocks""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "variables.tf" - file_path.write_text(''' + file_path.write_text( + """ variable "cloud_sql_version" { description = "Cloud SQL database version" default = "POSTGRES_12" } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) assert isinstance(findings, list) - + def test_multiple_databases_in_file(self): """Should detect deprecated versions in multiple Cloud SQL instances""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_sql_database_instance" "postgres" { database_version = "POSTGRES_10" } @@ -95,16 +106,18 @@ def test_multiple_databases_in_file(self): resource "google_sql_database_instance" "sql" { database_version = "SQLSERVER_2016" } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) assert isinstance(findings, list) - + def test_gcp_specific_fields(self): """Should detect GCP Cloud SQL-specific configuration""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "cloudsql.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_sql_database_instance" "primary" { name = "my-database" database_version = "POSTGRES_11" @@ -121,17 +134,19 @@ def test_gcp_specific_fields(self): } } } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) # PostgreSQL 11 is deprecated assert isinstance(findings, list) - + def test_detects_all_postgres_versions(self): """Should detect all PostgreSQL versions""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ # Old versions resource "google_sql_database_instance" "pg9" { database_version = "POSTGRES_9_6" @@ -144,16 +159,18 @@ def test_detects_all_postgres_versions(self): resource "google_sql_database_instance" "pg11" { database_version = "POSTGRES_11" } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) assert isinstance(findings, list) - + def test_detects_all_mysql_versions(self): """Should detect all MySQL versions""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_sql_database_instance" "mysql57" { database_version = "MYSQL_5_7" } @@ -161,7 +178,8 @@ def test_detects_all_mysql_versions(self): resource "google_sql_database_instance" "mysql80" { database_version = "MYSQL_8_0_31" } -''') - +""" + ) + findings = scan_gcp_cloudsql_versions(file_path) assert isinstance(findings, list) diff --git a/tests/test_gcp_detector.py b/tests/test_gcp_detector.py index 96dfc4b..475a46a 100644 --- a/tests/test_gcp_detector.py +++ b/tests/test_gcp_detector.py @@ -11,87 +11,98 @@ class TestGCPGKEVersionDetection: """Test Google Cloud GKE version detection and deprecation warnings""" - + def test_detects_deprecated_gke_version(self): """Should detect deprecated Kubernetes versions in GKE""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_container_cluster" "primary" { min_master_version = "1.24" } -''') - +""" + ) + findings = scan_gcp_versions(file_path) # Version 1.24 is deprecated on GCP assert isinstance(findings, list) - + def test_detects_rapid_release_channel_warning(self): """Should flag RAPID release channel (not recommended)""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_container_cluster" "primary" { release_channel { channel = "RAPID" } } -''') - +""" + ) + findings = scan_gcp_versions(file_path) # RAPID channel should be flagged assert isinstance(findings, list) - + def test_no_warning_for_stable_channel(self): """Should report STABLE channel with informational message""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_container_cluster" "primary" { release_channel { channel = "STABLE" } } -''') - +""" + ) + findings = scan_gcp_versions(file_path) # STABLE channel gets reported as informational assert len(findings) >= 1 assert "STABLE" in findings[0]["message"] - + def test_recommends_regular_over_rapid(self): """Should prefer REGULAR channel over RAPID""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_container_cluster" "primary" { release_channel { channel = "UNSPECIFIED" } } -''') - +""" + ) + findings = scan_gcp_versions(file_path) assert isinstance(findings, list) - + def test_detects_version_in_node_pool(self): """Should detect versions in node pool configuration""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_container_node_pool" "primary_nodes" { version = "1.25" } -''') - +""" + ) + findings = scan_gcp_versions(file_path) assert isinstance(findings, list) - + def test_multiple_gke_clusters(self): """Should detect deprecated versions in multiple GKE clusters""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_container_cluster" "primary" { min_master_version = "1.23" } @@ -99,16 +110,18 @@ def test_multiple_gke_clusters(self): resource "google_container_cluster" "secondary" { min_master_version = "1.24" } -''') - +""" + ) + findings = scan_gcp_versions(file_path) assert isinstance(findings, list) - + def test_gcp_specific_fields(self): """Should detect GCP-specific configuration patterns""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "gke.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "google_container_cluster" "primary" { name = "my-gke-cluster" location = "us-central1" @@ -121,7 +134,8 @@ def test_gcp_specific_fields(self): version = "1.22" } } -''') - +""" + ) + findings = scan_gcp_versions(file_path) assert isinstance(findings, list) diff --git a/tests/test_intelligent_detector.py b/tests/test_intelligent_detector.py index 7438d1e..1dc5620 100644 --- a/tests/test_intelligent_detector.py +++ b/tests/test_intelligent_detector.py @@ -9,20 +9,20 @@ class TestEntropyDetection: """Test entropy-based secret detection""" - + def test_detects_high_entropy_string(self): """High-entropy random strings should be detected as secrets""" line = 'password = "xK7mPqL9bJnR2tFhWdS4vE6cB8gA"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - assert findings[0]['confidence'] > 0.5 - + assert findings[0]["confidence"] > 0.5 + def test_ignores_low_entropy_word(self): """Common English words should NOT be flagged""" line = 'name = "production"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_entropy_calculation_works(self): """Entropy calculation should work correctly""" high_entropy = IntelligentDetector.calculate_entropy("xK7mPqL9bJnR2tFhWdS4vE6cB8gA") @@ -33,31 +33,31 @@ def test_entropy_calculation_works(self): class TestSemanticDetection: """Test semantic analysis (variable names indicating secrets)""" - + def test_detects_password_keyword(self): """Detects 'password' in variable name""" line = 'db_password = "SecureP@ss123456"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - + def test_detects_secret_keyword(self): """Detects 'secret' in variable name with high entropy""" line = 'api_secret = "xK7mPqL9bJnR2tFhWdS4vE6cB8gA"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - + def test_detects_token_keyword(self): """Detects 'token' in variable name""" line = 'auth_token = "bearer_xyz123456789abc"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - + def test_detects_api_key_keyword(self): """Detects 'api_key' in variable name with high entropy""" line = 'stripe_api_key = "xK7mPqL9bJnR2tFhWdS4vE6cB8gA"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - + def test_context_aware_detection(self): """Detects secrets using context from previous lines""" # Variable declaration 3 lines back @@ -69,34 +69,34 @@ def test_context_aware_detection(self): class TestFormatDetection: """Test detection of known secret formats (AWS, Stripe, GitHub, etc.)""" - + def test_detects_aws_access_key(self): """AWS Access Keys (AKIA prefix) should be detected with 95% confidence""" line = 'aws_key = "AKIAIOSFODNN7EXAMPLE"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - assert findings[0]['confidence'] >= 0.95 - + assert findings[0]["confidence"] >= 0.95 + def test_detects_stripe_secret_key(self): """Stripe Secret Keys (sk_live prefix) should be detected with 95% confidence""" line = 'stripe_secret = "sk_live_4eC39HqLyjWDarhtT8B3mkEv"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - assert findings[0]['confidence'] >= 0.95 - + assert findings[0]["confidence"] >= 0.95 + def test_detects_github_token(self): """GitHub tokens (ghp prefix) should be detected with 95% confidence""" line = 'github_token = "ghp_16C7e42F292c6912E7710c838347Ae178B4a"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - assert findings[0]['confidence'] >= 0.95 - + assert findings[0]["confidence"] >= 0.95 + def test_detects_google_oauth_token(self): """Google OAuth tokens (ya29 prefix) should be detected""" line = 'google_token = "ya29.a0AfH6SMBx_abcdef123456"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - + def test_detects_private_key(self): """Private key with content should be detected""" line = 'private_key = "-----BEGIN RSA PRIVATE KEY-----"' @@ -106,116 +106,116 @@ def test_detects_private_key(self): class TestFalsePositiveExclusions: """Test that legitimate patterns are NOT flagged as secrets""" - + def test_excludes_terraform_interpolations(self): """Terraform variable interpolations should NOT be flagged""" line = 'Name = "${var.project_name}-public-subnet-${count.index + 1}"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_aws_arns(self): """AWS ARNs should NOT be flagged""" line = 'policy_arn = "arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_ec2_instance_id(self): """EC2 Instance IDs (i-*) should NOT be flagged""" line = 'instance_id = "i-1234567890abcdef0"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_security_group_id(self): """Security Group IDs (sg-*) should NOT be flagged""" line = 'security_group = "sg-123456789"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_vpc_id(self): """VPC IDs (vpc-*) should NOT be flagged""" line = 'vpc_id = "vpc-12345678"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_subnet_id(self): """Subnet IDs (subnet-*) should NOT be flagged""" line = 'subnet_id = "subnet-123456789"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_naming_patterns(self): """Resource naming patterns should NOT be flagged""" line = 'tag_name = "web-server-prod-us-east-1"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_resource_references(self): """Terraform resource references should NOT be flagged""" - line = 'db_endpoint = aws_db_instance.postgres.endpoint' + line = "db_endpoint = aws_db_instance.postgres.endpoint" findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_version_numbers(self): """Version numbers should NOT be flagged""" line = 'kubernetes_version = "1.27.0"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_container_images(self): """Container image names should NOT be flagged""" line = 'image = "nginx:1.24"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_docker_registries(self): """Docker registries should NOT be flagged""" line = 'image = "gcr.io/my-project/app:v1.2.3"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_urls(self): """URLs should NOT be flagged""" line = 'endpoint = "https://api.example.com/v1/users"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_ip_addresses(self): """IP addresses should NOT be flagged""" line = 'server_ip = "192.168.1.100"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_domain_names(self): """Domain names should NOT be flagged""" line = 'hostname = "api.example.com"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_email_addresses(self): """Email addresses should NOT be flagged""" line = 'email = "admin@example.com"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_kubernetes_patterns(self): """Kubernetes patterns should NOT be flagged""" - line = 'apiVersion: v1' + line = "apiVersion: v1" findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 - + def test_excludes_kubernetes_kind(self): """Kubernetes kind declarations should NOT be flagged""" - line = 'kind: Deployment' + line = "kind: Deployment" findings = IntelligentDetector.detect_in_line(line) assert len(findings) == 0 class TestDetectSecretsFunction: """Test the main detect_secrets() function""" - + def test_detects_secrets_in_text(self): """detect_secrets() should find secrets in multi-line text""" - text = ''' + text = """ variable "db_password" { default = "MySecureP@ssw0rd123456789" } @@ -223,38 +223,38 @@ def test_detects_secrets_in_text(self): variable "api_key" { default = "sk_live_4eC39HqLyjWDarhtT8B3mkEv" } -''' +""" findings = detect_secrets(text) assert len(findings) >= 2 - + def test_detects_secrets_with_line_numbers(self): """Findings should include line numbers""" text = '''password = "SecureP@ss123456789" other_var = "production" api_key = "sk_live_4eC39HqLyjWDarhtT8B3mkEv"''' findings = detect_secrets(text) - assert any(f['line'] == 1 for f in findings) - assert any(f['line'] == 3 for f in findings) - + assert any(f["line"] == 1 for f in findings) + assert any(f["line"] == 3 for f in findings) + def test_excludes_false_positives_in_text(self): """detect_secrets() should exclude false positives""" - text = ''' + text = """ Name = "${var.environment}-public-subnet-${count.index}" arn = "arn:aws:iam::123456789:role/MyRole" version = "1.27.0" image = "nginx:latest" -''' +""" findings = detect_secrets(text) assert len(findings) == 0 - + def test_confidence_threshold_filtering(self): """Should only return findings above confidence threshold""" text = 'password = "SecureP@ss123456789"' - + # Low threshold - should find it findings_low = detect_secrets(text, min_confidence=0.5) assert len(findings_low) > 0 - + # Very high threshold - might not find it findings_high = detect_secrets(text, min_confidence=0.99) # Password should have ~63% confidence, so might not pass 99% threshold @@ -262,55 +262,55 @@ def test_confidence_threshold_filtering(self): class TestConfidenceScoring: """Test confidence score calculation""" - + def test_high_confidence_for_known_formats(self): """Known secret formats should have 95% confidence""" line = 'stripe_key = "sk_live_4eC39HqLyjWDarhtT8B3mkEv"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - assert findings[0]['confidence'] >= 0.95 - + assert findings[0]["confidence"] >= 0.95 + def test_medium_confidence_for_entropy_secrets(self): """High-entropy strings should have 60-80% confidence""" line = 'password = "xK7mPqL9bJnR2tFhWdS4vE6cB8gA"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 # Should have decent confidence due to entropy + keyword - assert findings[0]['confidence'] > 0.5 - + assert findings[0]["confidence"] > 0.5 + def test_context_increases_confidence(self): """Context from variable names should increase confidence""" # With context from 'db_password' variable context = "variable db_password { type = string default = " line = ' default = "MySecureP@ssw0rd123456789"' findings = IntelligentDetector.detect_in_line(line, context=context) - + # Without context line_only = ' default = "MySecureP@ssw0rd123456789"' findings_no_context = IntelligentDetector.detect_in_line(line_only) - + # Confidence with context should be higher if len(findings) > 0 and len(findings_no_context) > 0: - assert findings[0]['confidence'] >= findings_no_context[0]['confidence'] + assert findings[0]["confidence"] >= findings_no_context[0]["confidence"] class TestDetectionMethods: """Test that detection methods are correctly identified""" - + def test_identifies_format_detection(self): """Should identify when format-based detection is used""" line = 'stripe_key = "sk_live_4eC39HqLyjWDarhtT8B3mkEv"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - assert 'Stripe' in findings[0]['detection_method'] - + assert "Stripe" in findings[0]["detection_method"] + def test_identifies_entropy_detection(self): """Should identify when entropy-based detection is used""" line = 'password = "xK7mPqL9bJnR2tFhWdS4vE6cB8gA"' findings = IntelligentDetector.detect_in_line(line) assert len(findings) > 0 - assert 'Entropy' in findings[0]['detection_method'] - + assert "Entropy" in findings[0]["detection_method"] + def test_identifies_context_detection(self): """Should identify when context analysis is used""" context = "variable db_password = " diff --git a/tests/test_patterns.py b/tests/test_patterns.py index 8052806..1648b6d 100644 --- a/tests/test_patterns.py +++ b/tests/test_patterns.py @@ -1,11 +1,12 @@ from shieldcommit.intelligent_detector import IntelligentDetector + def test_intelligent_detector_has_secret_keywords(): """Verify intelligent detector has secret keywords configured.""" assert len(IntelligentDetector.SECRET_KEYWORDS) > 0 - assert 'password' in IntelligentDetector.SECRET_KEYWORDS - assert 'api_key' in IntelligentDetector.SECRET_KEYWORDS - assert 'token' in IntelligentDetector.SECRET_KEYWORDS + assert "password" in IntelligentDetector.SECRET_KEYWORDS + assert "api_key" in IntelligentDetector.SECRET_KEYWORDS + assert "token" in IntelligentDetector.SECRET_KEYWORDS def test_intelligent_detector_has_secret_structures(): @@ -19,5 +20,5 @@ def test_intelligent_detector_has_secret_structures(): def test_intelligent_detector_has_exclusions(): """Verify intelligent detector has exclusion keywords.""" assert len(IntelligentDetector.EXCLUDE_KEYWORDS) > 0 - assert 'arn:' in IntelligentDetector.EXCLUDE_KEYWORDS - assert 'version:' in IntelligentDetector.EXCLUDE_KEYWORDS + assert "arn:" in IntelligentDetector.EXCLUDE_KEYWORDS + assert "version:" in IntelligentDetector.EXCLUDE_KEYWORDS diff --git a/tests/test_rds_detector.py b/tests/test_rds_detector.py index 4bfa229..a050e80 100644 --- a/tests/test_rds_detector.py +++ b/tests/test_rds_detector.py @@ -11,108 +11,121 @@ class TestRDSVersionDetection: """Test AWS RDS version detection and deprecation warnings""" - + def test_detects_deprecated_postgres_version(self): """Should detect deprecated PostgreSQL versions in RDS""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_db_instance" "postgres" { engine = "postgres" engine_version = "12" } -''') - +""" + ) + findings = scan_rds_versions(file_path) assert len(findings) > 0 # PostgreSQL 12 should be flagged as deprecated - + def test_detects_deprecated_mysql_version(self): """Should detect deprecated MySQL versions in RDS""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_db_instance" "mysql" { engine = "mysql" engine_version = "5.7" } -''') - +""" + ) + findings = scan_rds_versions(file_path) assert len(findings) > 0 - + def test_detects_deprecated_mariadb_version(self): """Should detect deprecated MariaDB versions in RDS""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_db_instance" "mariadb" { engine = "mariadb" engine_version = "10.3" } -''') - +""" + ) + findings = scan_rds_versions(file_path) assert len(findings) > 0 - + def test_no_warning_for_current_version(self): """Should NOT warn for current database versions""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_db_instance" "postgres" { engine = "postgres" engine_version = "16" } -''') - +""" + ) + findings = scan_rds_versions(file_path) assert len(findings) == 0 - + def test_finds_version_in_variable_block(self): """Should find database versions in variable blocks""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "variables.tf" - file_path.write_text(''' + file_path.write_text( + """ variable "db_version" { description = "Database engine version" default = "13" } -''') - +""" + ) + findings = scan_rds_versions(file_path) assert len(findings) > 0 - + def test_includes_line_numbers(self): """Should include line numbers in findings""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_db_instance" "db" { engine = "postgres" engine_version = "11" } -''') - +""" + ) + findings = scan_rds_versions(file_path) # Version 11 is deprecated - just verify test works assert isinstance(findings, list) - + def test_includes_file_path(self): """Should include file path in findings""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "database.tf" file_path.write_text('engine = "mariadb"\nengine_version = "10.4"') - + findings = scan_rds_versions(file_path) # Just verify scanning works without errors assert isinstance(findings, list) - + def test_multiple_databases_in_file(self): """Should detect deprecated versions in multiple databases""" with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "main.tf" - file_path.write_text(''' + file_path.write_text( + """ resource "aws_db_instance" "postgres" { engine = "postgres" engine_version = "12" @@ -122,8 +135,9 @@ def test_multiple_databases_in_file(self): engine = "mysql" engine_version = "5.7" } -''') - +""" + ) + findings = scan_rds_versions(file_path) # Both versions are deprecated - just verify scanning works assert isinstance(findings, list) diff --git a/tests/test_scanner.py b/tests/test_scanner.py index 503ce8d..92c9972 100644 --- a/tests/test_scanner.py +++ b/tests/test_scanner.py @@ -6,30 +6,31 @@ def test_scan_detects_aws_key(): """Test intelligent detection of AWS access keys.""" text = 'my_key = "AKIAAAAAAAAAAAAAAAAA"' - - with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + + with tempfile.NamedTemporaryFile(mode="w", suffix=".tf", delete=False) as f: f.write(text) f.flush() - + results = scan_file(Path(f.name)) - + # Should detect the AWS key assert len(results) > 0 - assert any('AWS' in finding.get('detection_method', '') or - finding.get('confidence', 0) > 0.5 - for finding in results) + assert any( + "AWS" in finding.get("detection_method", "") or finding.get("confidence", 0) > 0.5 + for finding in results + ) def test_scan_detects_high_entropy_secrets(): """Test intelligent detection of high-entropy strings.""" text = 'password = "aB3$Kx9@mL2pQ5vN8xR1yT4gH"' - - with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + + with tempfile.NamedTemporaryFile(mode="w", suffix=".tf", delete=False) as f: f.write(text) f.flush() - + results = scan_file(Path(f.name)) - + # Should detect high-entropy secret assert len(results) > 0 @@ -37,26 +38,28 @@ def test_scan_detects_high_entropy_secrets(): def test_scan_excludes_arns(): """Test that ARNs are excluded from detection.""" text = 'arn = "arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly"' - - with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + + with tempfile.NamedTemporaryFile(mode="w", suffix=".tf", delete=False) as f: f.write(text) f.flush() - + results = scan_file(Path(f.name)) - + # Should NOT detect ARNs - assert len(results) == 0 or all('arn:aws' not in r.get('matched_value', '') for r in results) + assert len(results) == 0 or all( + "arn:aws" not in r.get("matched_value", "") for r in results + ) def test_scan_detects_semantic_secrets(): """Test semantic detection using variable names.""" text = 'api_key = "randomstringvalue1234567890"' - - with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + + with tempfile.NamedTemporaryFile(mode="w", suffix=".tf", delete=False) as f: f.write(text) f.flush() - + results = scan_file(Path(f.name)) - + # Should detect based on semantic analysis assert len(results) > 0