From 36ade61180c0485f7901b3cc6a8d9a219af56741 Mon Sep 17 00:00:00 2001 From: krishna fattepurkar Date: Thu, 1 Jan 2026 15:14:22 +0530 Subject: [PATCH 1/4] chore: add CI, contributing guide, and PR template --- .github/pull_request_template.md | 12 ++++++++++++ .github/workflows/ci.yml | 19 +++++++++++++++++++ CONTRIBUTING.md | 17 +++++++++++++++++ 3 files changed, 48 insertions(+) create mode 100644 .github/pull_request_template.md create mode 100644 .github/workflows/ci.yml create mode 100644 CONTRIBUTING.md diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..3e83e0f --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,12 @@ +## Summary +What does this PR do? + +## Type +- [ ] Feature +- [ ] Fix +- [ ] Docs +- [ ] CI + +## Testing +- [ ] Tested locally +- [ ] Tests added diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..5cb7ab0 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,19 @@ +name: CI + +on: + pull_request: + branches: [ main ] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.10' + - run: | + pip install -e . + pip install pytest flake8 + - run: flake8 shieldcommit + - run: pytest diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..b0f6ea7 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,17 @@ +# Contributing to ShieldCommit + +## Branch Rules +- main → stable only +- feature/* → new features +- fix/* → bug fixes +- docs/* → documentation + +## Workflow +1. Fork repo +2. Create branch from main +3. Open PR to main +4. CI must pass + +## Run locally +pip install -e . +pytest From 838e5dbcc2c8394a3cc41f39a58f2da9c46bcd16 Mon Sep 17 00:00:00 2001 From: krishna fattepurkar Date: Thu, 1 Jan 2026 15:28:00 +0530 Subject: [PATCH 2/4] chore: add CI, contributing guide, and PR template --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5cb7ab0..557122a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,5 +15,5 @@ jobs: - run: | pip install -e . pip install pytest flake8 - - run: flake8 shieldcommit + - run: flake8 src - run: pytest From 8950b321cd5ff8a6f12a32c888df5372cb3b01da Mon Sep 17 00:00:00 2001 From: krishna fattepurkar Date: Thu, 1 Jan 2026 15:38:53 +0530 Subject: [PATCH 3/4] chore: add CI, contributing guide, and PR template --- .github/workflows/ci.yml | 37 ++++++++++++++++++++++++++++--------- 1 file changed, 28 insertions(+), 9 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 557122a..7665965 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,19 +1,38 @@ -name: CI +name: ShieldCommit CI on: pull_request: - branches: [ main ] + push: + branches: [main] jobs: test: runs-on: ubuntu-latest + steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 with: - python-version: '3.10' - - run: | + python-version: "3.10" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip pip install -e . - pip install pytest flake8 - - run: flake8 src - - run: pytest + pip install -r requirements.txt || true + + - name: Verify CLI loads + run: | + shieldcommit --help + + - name: Run tests (if available) + run: | + if [ -d tests ]; then + pip install pytest + pytest + else + echo "No tests directory, skipping tests" + fi From 98c31c9bcb219621d61f56d50c7c32ec1ab006cc Mon Sep 17 00:00:00 2001 From: krishna fattepurkar Date: Thu, 1 Jan 2026 15:46:20 +0530 Subject: [PATCH 4/4] fixing test code --- tests/test_aks_detector.py | 188 +++++++++--------- tests/test_azure_db_detector.py | 261 +++++++++++++------------ tests/test_eks_detector.py | 200 +++++++++---------- tests/test_gcp_db_detector.py | 337 +++++++++++++++++--------------- tests/test_gcp_detector.py | 261 ++++++++++++++----------- tests/test_integration_v02.py | 337 ++++++++++++++++++++++++++++++++ tests/test_patterns.py | 25 +-- tests/test_rds_detector.py | 247 +++++++++++++---------- tests/test_scanner.py | 62 +----- 9 files changed, 1152 insertions(+), 766 deletions(-) create mode 100644 tests/test_integration_v02.py diff --git a/tests/test_aks_detector.py b/tests/test_aks_detector.py index 31b2686..f3365da 100644 --- a/tests/test_aks_detector.py +++ b/tests/test_aks_detector.py @@ -1,100 +1,112 @@ -""" -Tests for AKS (Azure Kubernetes Service) version detection -Detects deprecated/extended support Kubernetes versions on Azure -""" - -import pytest -from src.shieldcommit.aks_detector import scan_aks_versions -from pathlib import Path +import unittest import tempfile +from pathlib import Path +from src.shieldcommit.aks_detector import ( + parse_terraform_aks_versions, + get_version_warning, + scan_aks_versions, + AKS_VERSIONS +) -class TestAKSVersionDetection: - """Test Azure AKS version detection and deprecation warnings""" +class TestAKSDetector(unittest.TestCase): - 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(''' -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_parse_terraform_aks_versions(self): + """Test parsing AKS versions from Terraform content.""" + content = ''' + resource "azurerm_kubernetes_cluster" "example" { + kubernetes_version = "1.27" + name = "example-aks" + } + ''' + findings = parse_terraform_aks_versions(content) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["version"], "1.27") + self.assertEqual(findings[0]["line"], 3) - 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(''' -resource "azurerm_kubernetes_cluster" "aks" { - kubernetes_version = "1.25" -} -''') - - findings = scan_aks_versions(file_path) - assert isinstance(findings, list) + def test_parse_multiple_versions(self): + """Test parsing multiple AKS versions.""" + content = ''' + resource "azurerm_kubernetes_cluster" "cluster1" { + kubernetes_version = "1.27" + } + + resource "azurerm_kubernetes_cluster" "cluster2" { + kubernetes_version = "1.30" + } + ''' + findings = parse_terraform_aks_versions(content) + self.assertEqual(len(findings), 2) + versions = [f["version"] for f in findings] + self.assertIn("1.27", versions) + self.assertIn("1.30", versions) - 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(''' -resource "azurerm_kubernetes_cluster" "main" { - kubernetes_version = "1.29" -} -''') - - findings = scan_aks_versions(file_path) - assert len(findings) == 0 + def test_get_version_warning_deprecated(self): + """Test warning message for deprecated version.""" + warning = get_version_warning("1.25") + self.assertIn("DEPRECATED", warning) + self.assertIn("🚨", warning) + self.assertIn("2024-11", warning) - 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(''' -variable "aks_version" { - description = "AKS cluster version" - default = "1.24" -} -''') + def test_get_version_warning_extended(self): + """Test warning message for extended support version.""" + warning = get_version_warning("1.27") + self.assertIn("Extended Support", warning) + self.assertIn("⚠️", warning) + self.assertIn("2025-07", warning) + + def test_get_version_warning_current(self): + """Test info message for current version.""" + warning = get_version_warning("1.30") + self.assertIn("currently supported", warning) + self.assertIn("✓", warning) + + def test_get_version_warning_unknown(self): + """Test warning for unknown version.""" + warning = get_version_warning("1.99") + self.assertIn("unknown", warning) + self.assertIn("⚠️", warning) + + def test_scan_aks_versions_tf_file(self): + """Test scanning a Terraform file for AKS warnings.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "cluster.tf" + tf_file.write_text(''' + resource "azurerm_kubernetes_cluster" "example" { + kubernetes_version = "1.25" + } + ''') - findings = scan_aks_versions(file_path) - # Version 1.24 is deprecated - assert isinstance(findings, list) + warnings = scan_aks_versions(tf_file) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["type"], "aks_version") + self.assertEqual(warnings[0]["status"], "deprecated") - 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(''' -resource "azurerm_kubernetes_cluster" "cluster1" { - kubernetes_version = "1.23" -} - -resource "azurerm_kubernetes_cluster" "cluster2" { - kubernetes_version = "1.24" -} -''') + def test_scan_aks_versions_current_version(self): + """Test that current versions don't trigger warnings.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "cluster.tf" + tf_file.write_text(''' + resource "azurerm_kubernetes_cluster" "example" { + kubernetes_version = "1.30" + } + ''') - findings = scan_aks_versions(file_path) - assert isinstance(findings, list) + warnings = scan_aks_versions(tf_file) + self.assertEqual(len(warnings), 0) - 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(''' -resource "azurerm_kubernetes_cluster" "main" { - name = "my-aks" - kubernetes_version = "1.22" - node_resource_group = "MC_rg_aks_eastus" -} -''') + def test_scan_aks_versions_non_tf_file(self): + """Test that non-Terraform files are ignored.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + py_file = tmp_path / "script.py" + py_file.write_text('kubernetes_version = "1.25"') - findings = scan_aks_versions(file_path) - assert isinstance(findings, list) + warnings = scan_aks_versions(py_file) + self.assertEqual(len(warnings), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_azure_db_detector.py b/tests/test_azure_db_detector.py index d9b4b43..f2218c2 100644 --- a/tests/test_azure_db_detector.py +++ b/tests/test_azure_db_detector.py @@ -1,135 +1,156 @@ -""" -Tests for Azure Database version detection -Detects deprecated database engine versions on Azure -""" - -import pytest -from src.shieldcommit.azure_db_detector import scan_azure_db_versions -from pathlib import Path +import unittest import tempfile +from pathlib import Path +from src.shieldcommit.azure_db_detector import ( + parse_terraform_azure_db_versions, + get_azure_db_warning, + scan_azure_db_versions, + AZURE_DB_ENGINES +) -class TestAzureDBVersionDetection: - """Test Azure Database version detection and deprecation warnings""" +class TestAzureDBDetector(unittest.TestCase): - 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(''' -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_parse_terraform_azure_mysql_versions(self): + """Test parsing Azure MySQL versions from Terraform content.""" + content = ''' + resource "azurerm_mysql_server" "example" { + version = "8.0" + name = "example-mysql" + } + ''' + findings = parse_terraform_azure_db_versions(content) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["engine"], "mysql") + self.assertEqual(findings[0]["version"], "8.0") - 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(''' -resource "azurerm_mysql_server" "primary" { - version = "5.7" -} -''') - - findings = scan_azure_db_versions(file_path) - assert isinstance(findings, list) + def test_parse_terraform_azure_postgresql_versions(self): + """Test parsing Azure PostgreSQL versions from Terraform content.""" + content = ''' + resource "azurerm_postgresql_server" "example" { + version = "11" + name = "example-postgres" + } + ''' + findings = parse_terraform_azure_db_versions(content) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["engine"], "postgres") + self.assertEqual(findings[0]["version"], "11") - 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(''' -resource "azurerm_postgresql_server" "primary" { - version = "10" -} -''') - - findings = scan_azure_db_versions(file_path) - assert isinstance(findings, list) + def test_parse_terraform_azure_sql_versions(self): + """Test parsing Azure SQL versions from Terraform content.""" + content = ''' + resource "azurerm_mssql_server" "example" { + name = "example-sql" + sku_name = "GP_Gen5_2" + } + ''' + findings = parse_terraform_azure_db_versions(content) + # SQL detection is via SKU + self.assertTrue(any(f["engine"] == "sql" for f in findings) or len(findings) == 0) - 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(''' -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_parse_multiple_azure_db_versions(self): + """Test parsing multiple Azure database versions.""" + content = ''' + resource "azurerm_mysql_server" "mysql1" { + version = "5.7" + } + + resource "azurerm_postgresql_server" "postgres1" { + version = "12" + } + ''' + findings = parse_terraform_azure_db_versions(content) + self.assertTrue(len(findings) >= 2) + + def test_get_azure_db_warning_deprecated_mysql(self): + """Test warning message for deprecated MySQL version.""" + warning = get_azure_db_warning("mysql", "5.7") + self.assertIn("DEPRECATED", warning) + self.assertIn("🚨", warning) + self.assertIn("2024-10", warning) - 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(''' -resource "azurerm_postgresql_flexible_server" "primary" { - version = "11" -} -''') + def test_get_azure_db_warning_deprecated_postgres(self): + """Test warning message for deprecated PostgreSQL version.""" + warning = get_azure_db_warning("postgres", "11") + self.assertIn("DEPRECATED", warning) + self.assertIn("🚨", warning) + self.assertIn("2023-10", warning) + + def test_get_azure_db_warning_extended_postgres(self): + """Test warning message for extended support PostgreSQL version.""" + warning = get_azure_db_warning("postgres", "13") + self.assertIn("Extended Support", warning) + self.assertIn("⚠️", warning) + self.assertIn("2025-11", warning) + + def test_get_azure_db_warning_current(self): + """Test info message for current version.""" + warning = get_azure_db_warning("mysql", "8.0") + self.assertIn("currently supported", warning) + self.assertIn("✓", warning) + + def test_get_azure_db_warning_unknown(self): + """Test warning for unknown version.""" + warning = get_azure_db_warning("mysql", "9.0") + self.assertIn("unknown", warning) + self.assertIn("⚠️", warning) + + def test_scan_azure_db_versions_deprecated(self): + """Test scanning a Terraform file for deprecated Azure database versions.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "database.tf" + tf_file.write_text(''' + resource "azurerm_mysql_server" "example" { + version = "5.7" + } + ''') - findings = scan_azure_db_versions(file_path) - assert isinstance(findings, list) + warnings = scan_azure_db_versions(tf_file) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["type"], "azure_db_version") + self.assertEqual(warnings[0]["status"], "deprecated") - 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(''' -variable "db_version" { - description = "Azure database version" - default = "11" -} -''') + def test_scan_azure_db_versions_current(self): + """Test that current versions don't trigger warnings.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "database.tf" + tf_file.write_text(''' + resource "azurerm_mysql_server" "example" { + version = "8.0" + } + ''') - findings = scan_azure_db_versions(file_path) - assert isinstance(findings, list) + warnings = scan_azure_db_versions(tf_file) + self.assertEqual(len(warnings), 0) - 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(''' -resource "azurerm_postgresql_server" "postgres" { - version = "10" -} - -resource "azurerm_mysql_server" "mysql" { - version = "5.7" -} - -resource "azurerm_mssql_server" "sql" { - version = "12.0" -} -''') + def test_scan_azure_db_versions_non_tf_file(self): + """Test that non-Terraform files are ignored.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + py_file = tmp_path / "script.py" + py_file.write_text('version = "5.7"') - findings = scan_azure_db_versions(file_path) - assert isinstance(findings, list) + warnings = scan_azure_db_versions(py_file) + self.assertEqual(len(warnings), 0) - 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(''' -resource "azurerm_postgresql_server" "primary" { - name = "my-postgres" - location = "East US" - version = "9.5" - storage_mb = 5120 - backup_retention_days = 7 - geo_redundant_backup_enabled = true -} -''') + def test_scan_azure_db_versions_extended_support(self): + """Test detecting extended support versions.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "database.tf" + tf_file.write_text(''' + resource "azurerm_postgresql_server" "example" { + version = "13" + } + ''') - findings = scan_azure_db_versions(file_path) - # PostgreSQL 9.5 is deprecated - assert isinstance(findings, list) + warnings = scan_azure_db_versions(tf_file) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["status"], "extended") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_eks_detector.py b/tests/test_eks_detector.py index 9b549a3..57d66a9 100644 --- a/tests/test_eks_detector.py +++ b/tests/test_eks_detector.py @@ -1,113 +1,115 @@ """ -Tests for EKS (Elastic Kubernetes Service) version detection -Detects deprecated/extended support Kubernetes versions +Tests for EKS detector module. """ import pytest -from src.shieldcommit.eks_detector import scan_eks_versions from pathlib import Path import tempfile +from shieldcommit.eks_detector import ( + parse_terraform_eks_versions, + get_version_warning, + scan_eks_versions +) -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(''' -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(''' +def test_parse_terraform_eks_versions(): + """Test parsing of EKS versions from Terraform content.""" + content = """ resource "aws_eks_cluster" "main" { + name = "my-cluster" kubernetes_version = "1.27" + role_arn = aws_iam_role.eks_cluster.arn } -''') - - 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(''' -resource "aws_eks_cluster" "main" { + +resource "aws_eks_cluster" "secondary" { 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(''' -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(''' -# 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 - - 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'] - - 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(''' -resource "aws_eks_cluster" "cluster1" { - kubernetes_version = "1.23" -} +""" + findings = parse_terraform_eks_versions(content) + assert len(findings) == 2 + assert findings[0]["version"] == "1.27" + assert findings[0]["line"] == 4 + assert findings[1]["version"] == "1.30" + assert findings[1]["line"] == 9 -resource "aws_eks_cluster" "cluster2" { - kubernetes_version = "1.24" -} -''') - - findings = scan_eks_versions(file_path) - assert len(findings) >= 2 + +def test_parse_terraform_eks_versions_no_match(): + """Test parsing when no EKS versions found.""" + content = "resource 'aws_s3_bucket' 'bucket' { }" + findings = parse_terraform_eks_versions(content) + assert len(findings) == 0 + + +def test_get_version_warning_current(): + """Test warning message for current EKS version.""" + msg = get_version_warning("1.30") + assert "currently supported" in msg + assert "1.30" in msg + + +def test_get_version_warning_extended(): + """Test warning message for extended support version.""" + msg = get_version_warning("1.27") + assert "Extended Support" in msg + assert "1.27" in msg + + +def test_get_version_warning_deprecated(): + """Test warning message for deprecated version.""" + msg = get_version_warning("1.25") + assert "DEPRECATED" in msg + assert "1.25" in msg + + +def test_get_version_warning_unknown(): + """Test warning for unknown version.""" + msg = get_version_warning("1.99") + assert "unknown" in msg + + +def test_scan_eks_versions_terraform_file(): + """Test scanning a Terraform file for EKS versions.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('kubernetes_version = "1.26"\n') + f.flush() + + path = Path(f.name) + warnings = scan_eks_versions(path) + + assert len(warnings) == 1 + assert warnings[0]["type"] == "eks_version" + assert warnings[0]["version"] == "1.26" + assert warnings[0]["status"] == "extended" + assert "Extended Support" in warnings[0]["message"] + + path.unlink() + + +def test_scan_eks_versions_non_terraform_file(): + """Test that non-Terraform files are skipped.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: + f.write('kubernetes_version = "1.26"\n') + f.flush() + + path = Path(f.name) + warnings = scan_eks_versions(path) + + assert len(warnings) == 0 + path.unlink() + + +def test_scan_eks_versions_current_version_no_warning(): + """Test that current versions don't generate warnings.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('kubernetes_version = "1.30"\n') + f.flush() + + path = Path(f.name) + warnings = scan_eks_versions(path) + + assert len(warnings) == 0 + path.unlink() + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_gcp_db_detector.py b/tests/test_gcp_db_detector.py index 167063b..d6fd945 100644 --- a/tests/test_gcp_db_detector.py +++ b/tests/test_gcp_db_detector.py @@ -1,167 +1,188 @@ -""" -Tests for GCP Cloud SQL version detection -Detects deprecated database engine versions on Google Cloud SQL -""" - -import pytest -from src.shieldcommit.gcp_db_detector import scan_gcp_cloudsql_versions -from pathlib import Path +import unittest import tempfile +from pathlib import Path +from src.shieldcommit.gcp_db_detector import ( + parse_terraform_gcp_cloudsql_versions, + parse_gcp_version_string, + get_gcp_cloudsql_warning, + scan_gcp_cloudsql_versions, + GCP_CLOUDSQL_ENGINES +) -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(''' -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(''' -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(''' -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(''' -resource "google_sql_database_instance" "primary" { - database_version = "POSTGRES_15" -} -''') +class TestGCPDatabaseDetector(unittest.TestCase): + + def test_parse_gcp_version_string_mysql(self): + """Test parsing GCP MySQL version strings.""" + engine, version = parse_gcp_version_string("MYSQL_8_0") + self.assertEqual(engine, "mysql") + self.assertEqual(version, "8.0") + + engine, version = parse_gcp_version_string("MYSQL_5_7") + self.assertEqual(engine, "mysql") + self.assertEqual(version, "5.7") + + def test_parse_gcp_version_string_postgres(self): + """Test parsing GCP PostgreSQL version strings.""" + engine, version = parse_gcp_version_string("POSTGRES_14") + self.assertEqual(engine, "postgres") + self.assertEqual(version, "14") + + engine, version = parse_gcp_version_string("POSTGRES_11") + self.assertEqual(engine, "postgres") + self.assertEqual(version, "11") + + def test_parse_gcp_version_string_sqlserver(self): + """Test parsing GCP SQL Server version strings.""" + engine, version = parse_gcp_version_string("SQLSERVER_2019") + self.assertEqual(engine, "sqlserver") + self.assertEqual(version, "2019") + + engine, version = parse_gcp_version_string("SQLSERVER_2017") + self.assertEqual(engine, "sqlserver") + self.assertEqual(version, "2017") + + def test_parse_terraform_gcp_cloudsql_mysql(self): + """Test parsing GCP Cloud SQL MySQL versions from Terraform content.""" + content = ''' + resource "google_sql_database_instance" "example" { + database_version = "MYSQL_8_0" + name = "example-mysql" + } + ''' + findings = parse_terraform_gcp_cloudsql_versions(content) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["engine"], "mysql") + self.assertEqual(findings[0]["version"], "8.0") + + def test_parse_terraform_gcp_cloudsql_postgres(self): + """Test parsing GCP Cloud SQL PostgreSQL versions from Terraform content.""" + content = ''' + resource "google_sql_database_instance" "example" { + database_version = "POSTGRES_15" + name = "example-postgres" + } + ''' + findings = parse_terraform_gcp_cloudsql_versions(content) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["engine"], "postgres") + self.assertEqual(findings[0]["version"], "15") + + def test_parse_terraform_gcp_cloudsql_sqlserver(self): + """Test parsing GCP Cloud SQL SQL Server versions from Terraform content.""" + content = ''' + resource "google_sql_database_instance" "example" { + database_version = "SQLSERVER_2019" + name = "example-sql" + } + ''' + findings = parse_terraform_gcp_cloudsql_versions(content) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["engine"], "sqlserver") + self.assertEqual(findings[0]["version"], "2019") + + def test_parse_multiple_gcp_cloudsql_versions(self): + """Test parsing multiple GCP Cloud SQL versions.""" + content = ''' + resource "google_sql_database_instance" "mysql" { + database_version = "MYSQL_5_7" + } + + resource "google_sql_database_instance" "postgres" { + database_version = "POSTGRES_12" + } + ''' + findings = parse_terraform_gcp_cloudsql_versions(content) + self.assertEqual(len(findings), 2) + + def test_get_gcp_cloudsql_warning_deprecated_mysql(self): + """Test warning message for deprecated MySQL version.""" + warning = get_gcp_cloudsql_warning("mysql", "5.7") + self.assertIn("DEPRECATED", warning) + self.assertIn("🚨", warning) + self.assertIn("2024-10", warning) + + def test_get_gcp_cloudsql_warning_deprecated_postgres(self): + """Test warning message for deprecated PostgreSQL version.""" + warning = get_gcp_cloudsql_warning("postgres", "11") + self.assertIn("DEPRECATED", warning) + self.assertIn("🚨", warning) + self.assertIn("2023-10", warning) + + def test_get_gcp_cloudsql_warning_extended_postgres(self): + """Test warning message for extended support PostgreSQL version.""" + warning = get_gcp_cloudsql_warning("postgres", "13") + self.assertIn("Extended Support", warning) + self.assertIn("⚠️", warning) + self.assertIn("2025-11", warning) + + def test_get_gcp_cloudsql_warning_current(self): + """Test info message for current version.""" + warning = get_gcp_cloudsql_warning("mysql", "8.0") + self.assertIn("currently supported", warning) + self.assertIn("✓", warning) + + def test_get_gcp_cloudsql_warning_unknown_version(self): + """Test warning for unknown version.""" + warning = get_gcp_cloudsql_warning("mysql", "9.0") + self.assertIn("unknown", warning) + self.assertIn("⚠️", warning) + + def test_scan_gcp_cloudsql_versions_deprecated(self): + """Test scanning a Terraform file for deprecated GCP Cloud SQL versions.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "cloudsql.tf" + tf_file.write_text(''' + resource "google_sql_database_instance" "example" { + database_version = "MYSQL_5_7" + } + ''') - 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(''' -variable "cloud_sql_version" { - description = "Cloud SQL database version" - default = "POSTGRES_12" -} -''') + warnings = scan_gcp_cloudsql_versions(tf_file) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["type"], "gcp_cloudsql_version") + self.assertEqual(warnings[0]["status"], "deprecated") + + def test_scan_gcp_cloudsql_versions_current(self): + """Test that current versions don't trigger warnings.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "cloudsql.tf" + tf_file.write_text(''' + resource "google_sql_database_instance" "example" { + database_version = "POSTGRES_16" + } + ''') - 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(''' -resource "google_sql_database_instance" "postgres" { - database_version = "POSTGRES_10" -} - -resource "google_sql_database_instance" "mysql" { - database_version = "MYSQL_5_7" -} - -resource "google_sql_database_instance" "sql" { - database_version = "SQLSERVER_2016" -} -''') + warnings = scan_gcp_cloudsql_versions(tf_file) + self.assertEqual(len(warnings), 0) + + def test_scan_gcp_cloudsql_versions_extended(self): + """Test detecting extended support versions.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "cloudsql.tf" + tf_file.write_text(''' + resource "google_sql_database_instance" "example" { + database_version = "POSTGRES_13" + } + ''') - 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(''' -resource "google_sql_database_instance" "primary" { - name = "my-database" - database_version = "POSTGRES_11" - deletion_protection = true - region = "us-central1" - - settings { - tier = "db-f1-micro" - availability_type = "REGIONAL" - backup_configuration { - enabled = true - start_time = "07:00" - transaction_log_retention_days = 7 - } - } -} -''') + warnings = scan_gcp_cloudsql_versions(tf_file) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["status"], "extended") + + def test_scan_gcp_cloudsql_versions_non_tf_file(self): + """Test that non-Terraform files are ignored.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + py_file = tmp_path / "script.py" + py_file.write_text('database_version = "MYSQL_5_7"') - 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(''' -# Old versions -resource "google_sql_database_instance" "pg9" { - database_version = "POSTGRES_9_6" -} + warnings = scan_gcp_cloudsql_versions(py_file) + self.assertEqual(len(warnings), 0) -resource "google_sql_database_instance" "pg10" { - database_version = "POSTGRES_10" -} -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(''' -resource "google_sql_database_instance" "mysql57" { - database_version = "MYSQL_5_7" -} - -resource "google_sql_database_instance" "mysql80" { - database_version = "MYSQL_8_0_31" -} -''') - - findings = scan_gcp_cloudsql_versions(file_path) - assert isinstance(findings, list) +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gcp_detector.py b/tests/test_gcp_detector.py index 96dfc4b..fb8bdd1 100644 --- a/tests/test_gcp_detector.py +++ b/tests/test_gcp_detector.py @@ -1,127 +1,162 @@ -""" -Tests for GCP GKE (Google Kubernetes Engine) version detection -Detects deprecated/extended support Kubernetes versions on Google Cloud -""" - -import pytest -from src.shieldcommit.gcp_detector import scan_gcp_versions -from pathlib import Path +import unittest import tempfile +from pathlib import Path +from src.shieldcommit.gcp_detector import ( + parse_terraform_gcp_versions, + get_version_warning, + get_channel_info, + scan_gcp_versions, + GCP_VERSIONS +) -class TestGCPGKEVersionDetection: - """Test Google Cloud GKE version detection and deprecation warnings""" +class TestGCPDetector(unittest.TestCase): - 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(''' -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_parse_terraform_gcp_versions_explicit(self): + """Test parsing GCP versions from Terraform content.""" + content = ''' + resource "google_container_cluster" "example" { + min_master_version = "1.27" + name = "example-gke" + } + ''' + findings = parse_terraform_gcp_versions(content) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["version"], "1.27") + self.assertEqual(findings[0]["line"], 3) + self.assertEqual(findings[0]["type"], "explicit") - 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(''' -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_parse_terraform_gcp_versions_channel(self): + """Test parsing GCP release channel from Terraform content.""" + content = ''' + resource "google_container_cluster" "example" { + release_channel { + channel = "REGULAR" + } + } + ''' + findings = parse_terraform_gcp_versions(content) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0]["channel"], "REGULAR") + self.assertEqual(findings[0]["type"], "channel") - 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(''' -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_parse_multiple_gcp_versions(self): + """Test parsing multiple GCP versions.""" + content = ''' + resource "google_container_cluster" "cluster1" { + min_master_version = "1.27" + } + + resource "google_container_cluster" "cluster2" { + min_master_version = "1.30" + } + ''' + findings = parse_terraform_gcp_versions(content) + self.assertEqual(len(findings), 2) + versions = [f["version"] for f in findings if f["type"] == "explicit"] + self.assertIn("1.27", versions) + self.assertIn("1.30", versions) + + def test_get_version_warning_deprecated(self): + """Test warning message for deprecated version.""" + warning = get_version_warning("1.25") + self.assertIn("DEPRECATED", warning) + self.assertIn("🚨", warning) + self.assertIn("2024-07", warning) + + def test_get_version_warning_extended(self): + """Test warning message for extended support version.""" + warning = get_version_warning("1.27") + self.assertIn("Extended Support", warning) + self.assertIn("⚠️", warning) + self.assertIn("2025-04", warning) - 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(''' -resource "google_container_cluster" "primary" { - release_channel { - channel = "UNSPECIFIED" - } -} -''') + def test_get_version_warning_current(self): + """Test info message for current version.""" + warning = get_version_warning("1.30") + self.assertIn("currently supported", warning) + self.assertIn("✓", warning) + + def test_get_version_warning_unknown(self): + """Test warning for unknown version.""" + warning = get_version_warning("1.99") + self.assertIn("unknown", warning) + self.assertIn("⚠️", warning) + + def test_get_channel_info_regular(self): + """Test channel info for REGULAR channel.""" + info = get_channel_info("REGULAR") + self.assertIn("REGULAR", info) + self.assertIn("production", info) + + def test_get_channel_info_rapid(self): + """Test channel info for RAPID channel.""" + info = get_channel_info("RAPID") + self.assertIn("RAPID", info) + + def test_get_channel_info_stable(self): + """Test channel info for STABLE channel.""" + info = get_channel_info("STABLE") + self.assertIn("STABLE", info) + + def test_scan_gcp_versions_deprecated(self): + """Test scanning a Terraform file for deprecated GCP versions.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "gke.tf" + tf_file.write_text(''' + resource "google_container_cluster" "example" { + min_master_version = "1.25" + } + ''') - findings = scan_gcp_versions(file_path) - assert isinstance(findings, list) + warnings = scan_gcp_versions(tf_file) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["type"], "gcp_version") + self.assertEqual(warnings[0]["status"], "deprecated") - 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(''' -resource "google_container_node_pool" "primary_nodes" { - version = "1.25" -} -''') + def test_scan_gcp_versions_current(self): + """Test that current versions don't trigger version warnings.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "gke.tf" + tf_file.write_text(''' + resource "google_container_cluster" "example" { + min_master_version = "1.30" + } + ''') - findings = scan_gcp_versions(file_path) - assert isinstance(findings, list) + warnings = scan_gcp_versions(tf_file) + self.assertEqual(len(warnings), 0) - 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(''' -resource "google_container_cluster" "primary" { - min_master_version = "1.23" -} - -resource "google_container_cluster" "secondary" { - min_master_version = "1.24" -} -''') + def test_scan_gcp_channel_warning(self): + """Test scanning for GCP release channel.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + tf_file = tmp_path / "gke.tf" + tf_file.write_text(''' + resource "google_container_cluster" "example" { + release_channel { + channel = "REGULAR" + } + } + ''') - findings = scan_gcp_versions(file_path) - assert isinstance(findings, list) + warnings = scan_gcp_versions(tf_file) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["type"], "gcp_channel") + self.assertEqual(warnings[0]["channel"], "REGULAR") - 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(''' -resource "google_container_cluster" "primary" { - name = "my-gke-cluster" - location = "us-central1" - - release_channel { - channel = "RAPID" - } - - node_pool { - version = "1.22" - } -} -''') + def test_scan_gcp_versions_non_tf_file(self): + """Test that non-Terraform files are ignored.""" + with tempfile.TemporaryDirectory() as tmp_path: + tmp_path = Path(tmp_path) + py_file = tmp_path / "script.py" + py_file.write_text('min_master_version = "1.25"') - findings = scan_gcp_versions(file_path) - assert isinstance(findings, list) + warnings = scan_gcp_versions(py_file) + self.assertEqual(len(warnings), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_integration_v02.py b/tests/test_integration_v02.py new file mode 100644 index 0000000..5960a2d --- /dev/null +++ b/tests/test_integration_v02.py @@ -0,0 +1,337 @@ +""" +Integration tests for ShieldCommit v0.2 - Testing secrets + version warnings together +Includes EKS, RDS, AKS, GCP versions + Azure and GCP database versions +""" + +import pytest +from pathlib import Path +import tempfile +from shieldcommit.scanner import scan_files + + +def test_scan_files_with_secrets_and_warnings(): + """Test scanning a file that has both secrets and version warnings.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "aws_eks_cluster" "main" { + kubernetes_version = "1.26" + secret_key = "sk_live_test1234567890abcdefghij" # This is a secret +} + +resource "aws_db_instance" "db" { + engine = "postgres" + engine_version = "12" + db_password = "password='super_secret_123'" # This is a secret +} +''') + f.flush() + + path = Path(f.name) + result = scan_files([str(path)]) + + # Verify result structure + assert "findings" in result + assert "warnings" in result + assert isinstance(result["findings"], list) + assert isinstance(result["warnings"], list) + + # Should have secrets (findings) + assert len(result["findings"]) > 0 + + # Should have version warnings + assert len(result["warnings"]) > 0 + + # Verify findings are secrets + secret_patterns = {f["pattern"] for f in result["findings"]} + assert "Stripe Secret Key" in secret_patterns or "Generic Password" in secret_patterns + + # Verify warnings are version issues + warning_types = {w["type"] for w in result["warnings"]} + assert "eks_version" in warning_types or "rds_version" in warning_types + + path.unlink() + + +def test_scan_files_warnings_only(): + """Test scanning a file that has only version warnings, no secrets.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "aws_eks_cluster" "main" { + kubernetes_version = "1.25" + name = "my-cluster" +} +''') + f.flush() + + path = Path(f.name) + result = scan_files([str(path)]) + + assert len(result["findings"]) == 0 + # Now detects both EKS and AKS warnings for the same version pattern + assert len(result["warnings"]) >= 1 + warning_types = {w["type"] for w in result["warnings"]} + assert "eks_version" in warning_types or "aks_version" in warning_types + + path.unlink() + + +def test_scan_files_secrets_only(): + """Test scanning a Python file with secrets (no version warnings).""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: + f.write(''' +api_key = "sk_live_test1234567890abcdefghij" +password = "password='secret123'" +''') + f.flush() + + path = Path(f.name) + result = scan_files([str(path)]) + + assert len(result["findings"]) > 0 + assert len(result["warnings"]) == 0 # Python files don't get version warnings + + path.unlink() + + +def test_scan_multiple_files(): + """Test scanning multiple files with mixed results.""" + files = [] + + # Create a Terraform file with warnings + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('kubernetes_version = "1.25"') + f.flush() + files.append(f.name) + + # Create a Python file with secrets + with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: + f.write('api_key = "sk_live_test1234567890abcdefghij"') + f.flush() + files.append(f.name) + + result = scan_files(files) + + assert len(result["findings"]) > 0 # From Python file + assert len(result["warnings"]) > 0 # From Terraform file + + for file in files: + Path(file).unlink() + + +def test_scan_aks_versions(): + """Test scanning for AKS version warnings.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "azurerm_kubernetes_cluster" "main" { + kubernetes_version = "1.26" + name = "my-aks-cluster" +} +''') + f.flush() + + path = Path(f.name) + result = scan_files([str(path)]) + + assert len(result["warnings"]) > 0 + warning_types = {w["type"] for w in result["warnings"]} + assert "aks_version" in warning_types + + # Check for extended support status + aks_warnings = [w for w in result["warnings"] if w["type"] == "aks_version"] + assert len(aks_warnings) > 0 + assert aks_warnings[0]["status"] == "extended" + + path.unlink() + + +def test_scan_gcp_versions(): + """Test scanning for GCP GKE version warnings.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "google_container_cluster" "main" { + min_master_version = "1.25" + name = "my-gke-cluster" +} +''') + f.flush() + + path = Path(f.name) + result = scan_files([str(path)]) + + assert len(result["warnings"]) > 0 + warning_types = {w["type"] for w in result["warnings"]} + assert "gcp_version" in warning_types + + # Check for deprecated status + gcp_warnings = [w for w in result["warnings"] if w["type"] == "gcp_version"] + assert len(gcp_warnings) > 0 + assert gcp_warnings[0]["status"] == "deprecated" + + path.unlink() + + +def test_scan_gcp_release_channel(): + """Test scanning for GCP release channel info.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "google_container_cluster" "main" { + release_channel { + channel = "RAPID" + } +} +''') + f.flush() + + path = Path(f.name) + result = scan_files([str(path)]) + + # Channel info is always reported (not filtered as warnings) + warning_types = {w["type"] for w in result["warnings"]} + if "gcp_channel" in warning_types: + gcp_channels = [w for w in result["warnings"] if w["type"] == "gcp_channel"] + assert len(gcp_channels) > 0 + assert gcp_channels[0]["channel"] == "RAPID" + + path.unlink() + + +def test_scan_all_cloud_providers(): + """Test scanning files from all four cloud providers.""" + files = [] + + # EKS + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('kubernetes_version = "1.26" # EKS') + f.flush() + files.append(f.name) + + # RDS + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +engine = "postgres" +engine_version = "12" +''') + f.flush() + files.append(f.name) + + # AKS + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('kubernetes_version = "1.25" # AKS') + f.flush() + files.append(f.name) + + # GCP GKE + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('min_master_version = "1.24" # GCP GKE') + f.flush() + files.append(f.name) + + result = scan_files(files) + + # Should have warnings from all providers + warning_types = {w["type"] for w in result["warnings"]} + assert len(result["warnings"]) > 0 + + for file in files: + Path(file).unlink() + + +def test_scan_azure_database_versions(): + """Test scanning for Azure database version warnings.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "azurerm_mysql_server" "example" { + version = "5.7" +} +''') + f.flush() + + path = Path(f.name) + result = scan_files([str(path)]) + + assert len(result["warnings"]) > 0 + warning_types = {w["type"] for w in result["warnings"]} + assert "azure_db_version" in warning_types + + azure_warnings = [w for w in result["warnings"] if w["type"] == "azure_db_version"] + assert len(azure_warnings) > 0 + assert azure_warnings[0]["status"] == "deprecated" + + path.unlink() + + +def test_scan_gcp_cloudsql_versions(): + """Test scanning for GCP Cloud SQL database version warnings.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "google_sql_database_instance" "example" { + database_version = "MYSQL_5_7" +} +''') + f.flush() + + path = Path(f.name) + result = scan_files([str(path)]) + + assert len(result["warnings"]) > 0 + warning_types = {w["type"] for w in result["warnings"]} + assert "gcp_cloudsql_version" in warning_types + + gcp_warnings = [w for w in result["warnings"] if w["type"] == "gcp_cloudsql_version"] + assert len(gcp_warnings) > 0 + assert gcp_warnings[0]["status"] == "deprecated" + + path.unlink() + + +def test_scan_all_platforms_and_databases(): + """Test scanning all cloud platforms and their databases.""" + files = [] + + # Kubernetes platforms + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('kubernetes_version = "1.25" # EKS/AKS') + f.flush() + files.append(f.name) + + # AWS RDS + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('engine_version = "12" # RDS') + f.flush() + files.append(f.name) + + # Azure Database + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "azurerm_mysql_server" "example" { + version = "5.7" +} +''') + f.flush() + files.append(f.name) + + # GCP Cloud SQL + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "google_sql_database_instance" "example" { + database_version = "POSTGRES_11" +} +''') + f.flush() + files.append(f.name) + + result = scan_files(files) + + # Should have warnings from all platforms and databases + assert len(result["warnings"]) > 0 + + warning_types = {w["type"] for w in result["warnings"]} + # Check for at least some of the expected warning types + assert len(warning_types) > 0 + + for file in files: + Path(file).unlink() + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) \ No newline at end of file diff --git a/tests/test_patterns.py b/tests/test_patterns.py index 8052806..15b878d 100644 --- a/tests/test_patterns.py +++ b/tests/test_patterns.py @@ -1,23 +1,4 @@ -from shieldcommit.intelligent_detector import IntelligentDetector +from shieldcommit.patterns import PATTERNS -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 - - -def test_intelligent_detector_has_secret_structures(): - """Verify intelligent detector has known secret formats.""" - assert len(IntelligentDetector.SECRET_STRUCTURES) > 0 - # Should have AWS, Stripe, GitHub, and other formats - format_count = len(IntelligentDetector.SECRET_STRUCTURES) - assert format_count >= 10 - - -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 +def test_patterns_not_empty(): + assert len(PATTERNS) > 0 diff --git a/tests/test_rds_detector.py b/tests/test_rds_detector.py index 4bfa229..9d89390 100644 --- a/tests/test_rds_detector.py +++ b/tests/test_rds_detector.py @@ -1,129 +1,162 @@ """ -Tests for RDS (Relational Database Service) version detection -Detects deprecated database engine versions +Tests for RDS detector module. """ import pytest -from src.shieldcommit.rds_detector import scan_rds_versions from pathlib import Path import tempfile +from shieldcommit.rds_detector import ( + parse_terraform_rds_versions, + get_rds_warning, + scan_rds_versions +) -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(''' +def test_parse_terraform_rds_versions(): + """Test parsing of RDS versions from Terraform content.""" + content = """ resource "aws_db_instance" "postgres" { - engine = "postgres" - engine_version = "12" + identifier = "my-db" + engine = "postgres" + engine_version = "13" } -''') - - 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(''' + resource "aws_db_instance" "mysql" { - engine = "mysql" - engine_version = "5.7" + engine = "mysql" + engine_version = "8.0" } -''') - - 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(''' +""" + findings = parse_terraform_rds_versions(content) + assert len(findings) == 2 + assert findings[0]["engine"] == "postgres" + assert findings[0]["version"] == "13" + assert findings[1]["engine"] == "mysql" + assert findings[1]["version"] == "8.0" + + +def test_parse_terraform_rds_versions_mariadb(): + """Test parsing MariaDB versions.""" + content = """ 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(''' -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(''' -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(''' -resource "aws_db_instance" "db" { - engine = "postgres" - engine_version = "11" + engine = "mariadb" + engine_version = "10.5" } -''') - - 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(''' -resource "aws_db_instance" "postgres" { +""" + findings = parse_terraform_rds_versions(content) + assert len(findings) == 1 + assert findings[0]["engine"] == "mariadb" + assert findings[0]["version"] == "10.5" + + +def test_parse_terraform_rds_versions_no_match(): + """Test parsing when no RDS versions found.""" + content = "resource 'aws_s3_bucket' 'bucket' { }" + findings = parse_terraform_rds_versions(content) + assert len(findings) == 0 + + +def test_get_rds_warning_current(): + """Test warning message for current RDS version.""" + msg = get_rds_warning("postgres", "15") + assert "currently supported" in msg + assert "15" in msg + + +def test_get_rds_warning_extended(): + """Test warning message for extended support version.""" + msg = get_rds_warning("postgres", "13") + assert "Extended Support" in msg + assert "13" in msg + + +def test_get_rds_warning_deprecated(): + """Test warning message for deprecated version.""" + msg = get_rds_warning("postgres", "12") + assert "DEPRECATED" in msg + assert "12" in msg + + +def test_get_rds_warning_unknown_engine(): + """Test warning for unknown engine.""" + msg = get_rds_warning("oracle", "19c") + assert "not recognized" in msg + + +def test_get_rds_warning_unknown_version(): + """Test warning for unknown version of known engine.""" + msg = get_rds_warning("postgres", "99.99") + assert "unknown" in msg + + +def test_scan_rds_versions_terraform_file(): + """Test scanning a Terraform file for RDS versions.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('engine = "mariadb"\nengine_version = "10.4"\n') + f.flush() + + path = Path(f.name) + warnings = scan_rds_versions(path) + + assert len(warnings) == 1 + assert warnings[0]["type"] == "rds_version" + assert warnings[0]["engine"] == "mariadb" + assert warnings[0]["version"] == "10.4" + assert warnings[0]["status"] == "extended" + assert "Extended Support" in warnings[0]["message"] + + path.unlink() + + +def test_scan_rds_versions_non_terraform_file(): + """Test that non-Terraform files are skipped.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: + f.write('engine = "postgres"\nengine_version = "13"\n') + f.flush() + + path = Path(f.name) + warnings = scan_rds_versions(path) + + assert len(warnings) == 0 + path.unlink() + + +def test_scan_rds_versions_current_version_no_warning(): + """Test that current versions don't generate warnings.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write('engine = "mysql"\nengine_version = "8.0"\n') + f.flush() + + path = Path(f.name) + warnings = scan_rds_versions(path) + + assert len(warnings) == 0 + path.unlink() + + +def test_scan_rds_versions_multiple_deprecations(): + """Test scanning multiple deprecated versions.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.tf', delete=False) as f: + f.write(''' +resource "aws_db_instance" "old_postgres" { engine = "postgres" engine_version = "12" } -resource "aws_db_instance" "mysql" { +resource "aws_db_instance" "old_mysql" { engine = "mysql" engine_version = "5.7" } ''') - - findings = scan_rds_versions(file_path) - # Both versions are deprecated - just verify scanning works - assert isinstance(findings, list) + f.flush() + + path = Path(f.name) + warnings = scan_rds_versions(path) + + assert len(warnings) == 2 + assert all(w["status"] == "deprecated" for w in warnings) + + path.unlink() + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_scanner.py b/tests/test_scanner.py index 503ce8d..59e0137 100644 --- a/tests/test_scanner.py +++ b/tests/test_scanner.py @@ -1,62 +1,6 @@ -from shieldcommit.scanner import scan_file -from pathlib import Path -import tempfile - +from shieldcommit.scanner import scan_text 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: - 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) - - -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: - f.write(text) - f.flush() - - results = scan_file(Path(f.name)) - - # Should detect high-entropy secret - assert len(results) > 0 - - -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: - 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) - - -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: - f.write(text) - f.flush() - - results = scan_file(Path(f.name)) - - # Should detect based on semantic analysis - assert len(results) > 0 + results = scan_text(text) + assert any(name == 'AWS Access Key ID' for name, _ in results)