prowler/tests/providers/iac/iac_provider_test.py

import json
import os
import tempfile
from unittest import mock
from unittest.mock import MagicMock, patch

import pytest

from prowler.lib.check.models import CheckReportIAC
from prowler.lib.network.ssrf import OutboundURLNotAllowedError
from prowler.providers.iac.exceptions.exceptions import (
    IacRepositoryCloneError,
    IacScanError,
)
from prowler.providers.iac.iac_provider import IacProvider
from tests.providers.iac.iac_fixtures import (
    DEFAULT_SCAN_PATH,
    SAMPLE_ANOTHER_FAILED_CHECK,
    SAMPLE_ANOTHER_PASSED_CHECK,
    SAMPLE_ANOTHER_SKIPPED_CHECK,
    SAMPLE_CLOUDFORMATION_CHECK,
    SAMPLE_DOCKERFILE_CHECK,
    SAMPLE_FAILED_CHECK,
    SAMPLE_HIGH_SEVERITY_CHECK,
    SAMPLE_KUBERNETES_CHECK,
    SAMPLE_PASSED_CHECK,
    SAMPLE_SKIPPED_CHECK,
    SAMPLE_TRIVY_NON_CVE_VULNERABILITY,
    SAMPLE_TRIVY_VULNERABILITY_WITHOUT_CVE_ORG_REFERENCE,
    SAMPLE_TRIVY_VULNERABILITY_WITHOUT_REFERENCES,
    SAMPLE_YAML_CHECK,
    get_empty_trivy_output,
    get_invalid_trivy_output,
    get_sample_trivy_json_output,
)


class TestIacProvider:
    def test_iac_provider(self):
        """Test IAC provider with default parameters"""

        provider = IacProvider()

        assert provider._type == "iac"
        assert provider.type == "iac"
        assert provider.scan_path == DEFAULT_SCAN_PATH
        assert provider._audit_config == {}
        assert provider._mutelist is None

    def test_iac_provider_custom_scan_path(self):
        """Test IAC provider with custom scan path"""
        custom_path = "/custom/path"

        provider = IacProvider(scan_path=custom_path)

        assert provider._type == "iac"
        assert provider.scan_path == custom_path

    def test_iac_provider_process_finding_failed(self):
        """Test processing a failed finding"""
        provider = IacProvider()

        report = provider._process_finding(SAMPLE_FAILED_CHECK, "main.tf", "terraform")

        assert isinstance(report, CheckReportIAC)
        assert report.status == "FAIL"

        # Trivy emits "AVD-AWS-0001"; Hub indexes it without the AVD- prefix.
        expected_url = "https://hub.prowler.com/check/AWS-0001"
        assert report.check_metadata.Provider == "iac"
        assert report.check_metadata.CheckID == SAMPLE_FAILED_CHECK["ID"]
        assert report.check_metadata.CheckTitle == SAMPLE_FAILED_CHECK["Title"]
        assert report.check_metadata.Severity == "low"
        assert report.check_metadata.Remediation.Recommendation.Url == expected_url
        assert report.check_metadata.RelatedUrl == ""
        assert report.check_metadata.AdditionalURLs == [expected_url]

    def test_iac_provider_process_finding_passed(self):
        """Test processing a passed finding"""
        provider = IacProvider()

        report = provider._process_finding(SAMPLE_PASSED_CHECK, "main.tf", "terraform")

        assert isinstance(report, CheckReportIAC)
        assert report.status == "FAIL"  # Trivy findings are always FAIL by default

        assert report.check_metadata.Provider == "iac"
        assert report.check_metadata.CheckID == SAMPLE_PASSED_CHECK["ID"]
        assert report.check_metadata.CheckTitle == SAMPLE_PASSED_CHECK["Title"]
        assert report.check_metadata.Severity == "low"

    def test_iac_provider_process_vulnerability_prefers_cve_reference_and_filters_aqua(
        self,
    ):
        """Test CVE findings use cve.org and exclude Aqua references."""
        provider = IacProvider()

        report = provider._process_finding(
            {
                "VulnerabilityID": "CVE-2023-1234",
                "Title": "Example vulnerability",
                "Description": "This is an example vulnerability",
                "Severity": "high",
                "PrimaryURL": "https://avd.aquasec.com/nvd/cve-2023-1234",
                "References": [
                    "https://avd.aquasec.com/nvd/cve-2023-1234",
                    "https://nvd.nist.gov/vuln/detail/CVE-2023-1234",
                    "https://www.cve.org/CVERecord?id=CVE-2023-1234",
                    "https://security.example.com/advisories/CVE-2023-1234",
                ],
            },
            "package.json",
            "nodejs",
        )

        assert (
            report.check_metadata.Remediation.Recommendation.Url
            == "https://www.cve.org/CVERecord?id=CVE-2023-1234"
        )
        assert report.check_metadata.RelatedUrl == ""
        assert report.check_metadata.AdditionalURLs == [
            "https://www.cve.org/CVERecord?id=CVE-2023-1234"
        ]

    def test_iac_provider_process_vulnerability_builds_cve_org_for_nvd_reference(
        self,
    ):
        """Test official CVE URL is built when only NVD is provided."""
        provider = IacProvider()

        report = provider._process_finding(
            SAMPLE_TRIVY_VULNERABILITY_WITHOUT_CVE_ORG_REFERENCE,
            "package.json",
            "nodejs",
        )

        assert (
            report.check_metadata.Remediation.Recommendation.Url
            == "https://www.cve.org/CVERecord?id=CVE-2023-5678"
        )
        assert report.check_metadata.RelatedUrl == ""
        assert report.check_metadata.AdditionalURLs == [
            "https://www.cve.org/CVERecord?id=CVE-2023-5678"
        ]

    def test_iac_provider_process_vulnerability_builds_cve_org_when_references_missing(
        self,
    ):
        """Test CVE URL is built from VulnerabilityID when references are absent."""
        provider = IacProvider()

        report = provider._process_finding(
            SAMPLE_TRIVY_VULNERABILITY_WITHOUT_REFERENCES,
            "package.json",
            "nodejs",
        )

        assert (
            report.check_metadata.Remediation.Recommendation.Url
            == "https://www.cve.org/CVERecord?id=CVE-2023-9012"
        )
        assert report.check_metadata.RelatedUrl == ""
        assert report.check_metadata.AdditionalURLs == [
            "https://www.cve.org/CVERecord?id=CVE-2023-9012"
        ]

    def test_iac_provider_process_non_cve_vulnerability_falls_back_to_github_advisory(
        self,
    ):
        """Non-CVE vulnerabilities (GHSA-…) point to GitHub Security Advisories."""
        provider = IacProvider()

        report = provider._process_finding(
            SAMPLE_TRIVY_NON_CVE_VULNERABILITY,
            "package.json",
            "nodejs",
        )

        expected_url = (
            "https://github.com/advisories/"
            f"{SAMPLE_TRIVY_NON_CVE_VULNERABILITY['VulnerabilityID'].upper()}"
        )
        assert report.check_metadata.Remediation.Recommendation.Url == expected_url
        assert report.check_metadata.RelatedUrl == ""
        assert report.check_metadata.AdditionalURLs == [expected_url]

    @patch("subprocess.run")
    def test_iac_provider_run_scan_success(self, mock_subprocess):
        """Test successful IAC scan with Trivy"""
        provider = IacProvider()

        mock_subprocess.return_value = MagicMock(
            stdout=get_sample_trivy_json_output(), stderr=""
        )

        # Collect all batches from the generator
        reports = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], []
        ):
            reports.extend(batch)

        # Should have 3 misconfigurations from the sample output
        assert len(reports) == 3

        # Check that we have failed reports (Trivy findings are always FAIL by default)
        failed_reports = [r for r in reports if r.status == "FAIL"]
        assert len(failed_reports) == 3

        # Verify subprocess was called correctly
        mock_subprocess.assert_called_once_with(
            [
                "trivy",
                "fs",
                "/test/directory",
                "--format",
                "json",
                "--scanners",
                "vuln,misconfig,secret",
                "--parallel",
                "0",
                "--include-non-failures",
            ],
            capture_output=True,
            text=True,
        )

    @patch("subprocess.run")
    def test_iac_provider_run_scan_empty_output(self, mock_subprocess):
        """Test IAC scan with empty Trivy output"""
        provider = IacProvider()

        mock_subprocess.return_value = MagicMock(
            stdout=get_empty_trivy_output(), stderr=""
        )

        reports = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], []
        ):
            reports.extend(batch)
        assert len(reports) == 0

    def test_provider_run_local_scan(self):
        scan_path = "."
        provider = IacProvider(scan_path=scan_path)
        with mock.patch(
            "prowler.providers.iac.iac_provider.IacProvider.run_scan",
        ) as mock_run_scan:
            provider.run()
            mock_run_scan.assert_called_with(
                scan_path, ["vuln", "misconfig", "secret"], []
            )

    @mock.patch.dict(os.environ, {}, clear=True)
    def test_provider_run_remote_scan(self):
        scan_repository_url = "https://github.com/user/repo"
        with tempfile.TemporaryDirectory() as temp_dir:
            with (
                mock.patch(
                    "prowler.providers.iac.iac_provider.IacProvider._clone_repository",
                    return_value=(temp_dir, "main"),
                ) as mock_clone,
                mock.patch(
                    "prowler.providers.iac.iac_provider.IacProvider.run_scan"
                ) as mock_run_scan,
            ):
                # Repository cloning now happens during __init__
                provider = IacProvider(scan_repository_url=scan_repository_url)

                # Verify clone was called during initialization
                mock_clone.assert_called_once_with(
                    scan_repository_url, None, None, None
                )

                # Verify region was updated with branch name
                assert provider.region == "main"

                # Run the scan
                provider.run()

                # Verify scan was called with the cloned directory
                mock_run_scan.assert_called_with(
                    temp_dir, ["vuln", "misconfig", "secret"], []
                )

    @mock.patch.dict(os.environ, {}, clear=True)
    def test_print_credentials_local(self):
        scan_path = "/path/to/scan"
        provider = IacProvider(scan_path=scan_path)
        with mock.patch("builtins.print") as mock_print:
            provider.print_credentials()
            assert any(
                f"Directory: \x1b[33m{scan_path}\x1b[0m" in call.args[0]
                for call in mock_print.call_args_list
            )
            assert any(
                "Scanning local IaC directory:" in call.args[0]
                for call in mock_print.call_args_list
            )

    @mock.patch.dict(os.environ, {}, clear=True)
    def test_print_credentials_remote(self):
        repo_url = "https://github.com/user/repo"
        with tempfile.TemporaryDirectory() as temp_dir:
            with mock.patch(
                "prowler.providers.iac.iac_provider.IacProvider._clone_repository",
                return_value=(temp_dir, "main"),
            ):
                provider = IacProvider(scan_repository_url=repo_url)
                with mock.patch("builtins.print") as mock_print:
                    provider.print_credentials()
                    assert any(
                        f"Repository: \x1b[33m{repo_url}\x1b[0m" in call.args[0]
                        for call in mock_print.call_args_list
                    )
                    assert any(
                        "Scanning remote IaC repository:" in call.args[0]
                        for call in mock_print.call_args_list
                    )

    @patch("subprocess.run")
    def test_iac_provider_process_check_medium_severity(self, mock_subprocess):
        """Test processing a medium severity check"""
        provider = IacProvider()

        mock_subprocess.return_value = MagicMock(
            stdout=get_invalid_trivy_output(), stderr=""
        )

        with pytest.raises(IacScanError):
            # Consume the generator
            for _ in provider.run_scan("/test/directory", ["all"], []):
                pass

    @patch("subprocess.run")
    def test_iac_provider_run_scan_null_output(self, mock_subprocess):
        """Test IAC scan with null Trivy output"""
        provider = IacProvider()

        mock_subprocess.return_value = MagicMock(stdout="null", stderr="")

        with pytest.raises(IacScanError):
            # Consume the generator
            for _ in provider.run_scan(
                "/test/directory", ["vuln", "misconfig", "secret"], []
            ):
                pass

    def test_iac_provider_process_finding_dockerfile(self):
        """Test processing a Dockerfile finding"""
        provider = IacProvider()

        report = provider._process_finding(
            SAMPLE_DOCKERFILE_CHECK, "Dockerfile", "dockerfile"
        )

        assert isinstance(report, CheckReportIAC)
        assert report.status == "FAIL"
        assert report.check_metadata.ServiceName == "dockerfile"
        assert report.check_metadata.CheckID == SAMPLE_DOCKERFILE_CHECK["ID"]

    def test_iac_provider_process_finding_yaml(self):
        """Test processing a YAML finding"""
        provider = IacProvider()

        report = provider._process_finding(
            SAMPLE_YAML_CHECK, "deployment.yaml", "kubernetes"
        )

        assert isinstance(report, CheckReportIAC)
        assert report.status == "FAIL"  # Trivy findings are always FAIL by default
        assert report.check_metadata.ServiceName == "kubernetes"
        assert report.check_metadata.CheckID == SAMPLE_YAML_CHECK["ID"]

    @patch("subprocess.run")
    def test_run_scan_success_with_failed_and_passed_checks(self, mock_subprocess):
        """Test successful run_scan with both failed and passed checks"""
        provider = IacProvider()

        # Create sample Trivy output with both failed and passed checks
        sample_output = {
            "Results": [
                {
                    "Target": "main.tf",
                    "Type": "terraform",
                    "Misconfigurations": [SAMPLE_FAILED_CHECK, SAMPLE_PASSED_CHECK],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                }
            ]
        }

        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps(sample_output), stderr=""
        )

        result = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], []
        ):
            result.extend(batch)

        # Verify results
        assert len(result) == 2
        assert all(isinstance(report, CheckReportIAC) for report in result)

        # Check that we have FAIL reports (Trivy findings are always FAIL by default)
        statuses = [report.status for report in result]
        assert all(status == "FAIL" for status in statuses)

    @patch("subprocess.run")
    def test_run_scan_with_skipped_checks(self, mock_subprocess):
        """Test run_scan with skipped checks (muted)"""
        provider = IacProvider()

        # Create sample Trivy output with skipped checks
        sample_output = {
            "Results": [
                {
                    "Target": "main.tf",
                    "Type": "terraform",
                    "Misconfigurations": [SAMPLE_SKIPPED_CHECK],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                }
            ]
        }

        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps(sample_output), stderr=""
        )

        result = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], ["exclude/path"]
        ):
            result.extend(batch)

        # Verify results
        assert len(result) == 1
        assert isinstance(result[0], CheckReportIAC)
        assert result[0].status == "MUTED"
        assert result[0].muted is True

    @patch("subprocess.run")
    def test_run_scan_empty_results(self, mock_subprocess):
        """Test run_scan with no findings"""
        provider = IacProvider()

        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps({"Results": []}), stderr=""
        )

        result = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], []
        ):
            result.extend(batch)

        # Verify results
        assert len(result) == 0

    @patch("subprocess.run")
    def test_run_scan_multiple_reports(self, mock_subprocess):
        """Test run_scan with multiple reports from different frameworks"""
        provider = IacProvider()

        # Create sample Trivy output with multiple frameworks
        sample_output = {
            "Results": [
                {
                    "Target": "main.tf",
                    "Type": "terraform",
                    "Misconfigurations": [SAMPLE_FAILED_CHECK],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                },
                {
                    "Target": "deployment.yaml",
                    "Type": "kubernetes",
                    "Misconfigurations": [SAMPLE_PASSED_CHECK],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                },
            ]
        }

        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps(sample_output), stderr=""
        )

        result = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], []
        ):
            result.extend(batch)

        # Verify results
        assert len(result) == 2
        assert all(isinstance(report, CheckReportIAC) for report in result)

        # Check that we have FAIL reports (Trivy findings are always FAIL by default)
        statuses = [report.status for report in result]
        assert all(status == "FAIL" for status in statuses)

    @patch("subprocess.run")
    def test_run_scan_exception_handling(self, mock_subprocess):
        """Test run_scan exception handling"""
        provider = IacProvider()

        # Make subprocess.run raise an exception
        mock_subprocess.side_effect = Exception("Test exception")

        with pytest.raises(IacScanError):
            # Consume the generator
            for _ in provider.run_scan(
                "/test/directory", ["vuln", "misconfig", "secret"], []
            ):
                pass

    @patch("subprocess.run")
    def test_run_scan_with_different_frameworks(self, mock_subprocess):
        """Test run_scan with different scanner configurations"""
        provider = IacProvider()

        sample_output = {
            "Results": [
                {
                    "Target": "main.tf",
                    "Type": "terraform",
                    "Misconfigurations": [SAMPLE_PASSED_CHECK],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                }
            ]
        }

        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps(sample_output), stderr=""
        )

        # Test with specific scanners
        scanners = ["vuln", "misconfig", "secret"]
        result = []
        for batch in provider.run_scan("/test/directory", scanners, []):
            result.extend(batch)

        # Verify subprocess was called with correct scanners
        mock_subprocess.assert_called_once_with(
            [
                "trivy",
                "fs",
                "/test/directory",
                "--format",
                "json",
                "--scanners",
                ",".join(scanners),
                "--parallel",
                "0",
                "--include-non-failures",
            ],
            capture_output=True,
            text=True,
        )

        # Verify results
        assert len(result) == 1
        assert result[0].status == "FAIL"  # Trivy findings are always FAIL by default

    @patch("subprocess.run")
    def test_run_scan_with_exclude_paths(self, mock_subprocess):
        """Test run_scan with exclude paths"""
        provider = IacProvider()

        sample_output = {
            "Results": [
                {
                    "Target": "main.tf",
                    "Type": "terraform",
                    "Misconfigurations": [SAMPLE_PASSED_CHECK],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                }
            ]
        }

        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps(sample_output), stderr=""
        )

        # Test with exclude paths
        exclude_paths = ["node_modules", ".git", "vendor"]
        result = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], exclude_paths
        ):
            result.extend(batch)

        # Verify subprocess was called with correct exclude paths
        expected_command = [
            "trivy",
            "fs",
            "/test/directory",
            "--format",
            "json",
            "--scanners",
            "vuln,misconfig,secret",
            "--parallel",
            "0",
            "--include-non-failures",
            "--skip-dirs",
            ",".join(exclude_paths),
        ]
        mock_subprocess.assert_called_once_with(
            expected_command,
            capture_output=True,
            text=True,
        )

        # Verify results
        assert len(result) == 1
        assert result[0].status == "FAIL"  # Trivy findings are always FAIL by default

    @patch("subprocess.run")
    def test_run_scan_all_check_types(self, mock_subprocess):
        """Test run_scan with all types of checks (failed, passed, skipped)"""
        provider = IacProvider()

        sample_output = {
            "Results": [
                {
                    "Target": "main.tf",
                    "Type": "terraform",
                    "Misconfigurations": [
                        SAMPLE_FAILED_CHECK,
                        SAMPLE_HIGH_SEVERITY_CHECK,
                        SAMPLE_PASSED_CHECK,
                        SAMPLE_CLOUDFORMATION_CHECK,
                        SAMPLE_SKIPPED_CHECK,
                    ],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                }
            ]
        }

        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps(sample_output), stderr=""
        )

        # Consume the generator to get all batches
        result = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], []
        ):
            result.extend(batch)

        # Verify results
        assert len(result) == 5  # 5 misconfigurations

        # Check status distribution
        statuses = [report.status for report in result]
        assert statuses.count("FAIL") == 4  # 4 regular findings
        assert statuses.count("MUTED") == 1  # 1 skipped finding

        # Check that muted reports have muted=True
        muted_reports = [report for report in result if report.status == "MUTED"]
        assert all(report.muted for report in muted_reports)

    @patch("subprocess.run")
    def test_run_scan_no_reports_returned(self, mock_subprocess):
        """Test run_scan when no reports are returned from registry"""
        provider = IacProvider()

        # Return empty list of reports
        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps({"Results": []}), stderr=""
        )

        # Consume the generator to get all batches
        result = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], []
        ):
            result.extend(batch)

        # Verify results
        assert len(result) == 0

    @patch("subprocess.run")
    def test_run_scan_multiple_frameworks_with_different_checks(self, mock_subprocess):
        """Test run_scan with multiple frameworks and different types of checks"""
        provider = IacProvider()

        # Create sample Trivy output with multiple frameworks and different check types
        sample_output = {
            "Results": [
                {
                    "Target": "main.tf",
                    "Type": "terraform",
                    "Misconfigurations": [
                        SAMPLE_FAILED_CHECK,
                        SAMPLE_ANOTHER_FAILED_CHECK,
                        SAMPLE_PASSED_CHECK,
                    ],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                },
                {
                    "Target": "deployment.yaml",
                    "Type": "kubernetes",
                    "Misconfigurations": [
                        SAMPLE_KUBERNETES_CHECK,
                        SAMPLE_ANOTHER_SKIPPED_CHECK,
                    ],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                },
                {
                    "Target": "template.yaml",
                    "Type": "cloudformation",
                    "Misconfigurations": [
                        SAMPLE_CLOUDFORMATION_CHECK,
                        SAMPLE_ANOTHER_PASSED_CHECK,
                    ],
                    "Vulnerabilities": [],
                    "Secrets": [],
                    "Licenses": [],
                },
            ]
        }

        mock_subprocess.return_value = MagicMock(
            stdout=json.dumps(sample_output), stderr=""
        )

        # Consume the generator to get all batches
        result = []
        for batch in provider.run_scan(
            "/test/directory", ["vuln", "misconfig", "secret"], []
        ):
            result.extend(batch)

        # Verify results
        assert (
            len(result) == 7
        )  # 3 terraform + 2 kubernetes + 2 cloudformation = 7 total

        # Check status distribution
        statuses = [report.status for report in result]
        assert statuses.count("FAIL") == 6  # 6 regular findings
        assert statuses.count("MUTED") == 1  # 1 skipped finding

    def test_run_method_calls_run_scan(self):
        """Test that the run method calls run_scan with correct parameters"""
        provider = IacProvider(
            scan_path="/custom/path",
            scanners=["vuln", "misconfig"],
            exclude_path=["exclude"],
        )

        with patch.object(provider, "run_scan") as mock_run_scan:
            # Mock should return a generator (empty in this case)
            mock_run_scan.return_value = iter([])
            provider.run()

            mock_run_scan.assert_called_once_with(
                "/custom/path", ["vuln", "misconfig"], ["exclude"]
            )

    @mock.patch("prowler.providers.iac.iac_provider.porcelain.clone")
    @mock.patch("tempfile.mkdtemp", return_value="/tmp/fake-dir")
    def test_clone_repository_no_auth(self, _mock_mkdtemp, mock_clone):
        provider = IacProvider()
        url = "https://github.com/user/repo.git"
        with mock.patch.object(provider, "_detect_branch_name", return_value="main"):
            temp_dir, branch_name = provider._clone_repository(url)
        mock_clone.assert_called_with(
            url, "/tmp/fake-dir", depth=1, config=mock.ANY, pool_manager=mock.ANY
        )
        assert temp_dir == "/tmp/fake-dir"
        assert branch_name == "main"

    @mock.patch("prowler.providers.iac.iac_provider.porcelain.clone")
    @mock.patch("tempfile.mkdtemp", return_value="/tmp/fake-dir")
    def test_clone_repository_with_pat(self, _mock_mkdtemp, mock_clone):
        provider = IacProvider()
        url = "https://github.com/user/repo.git"
        with mock.patch.object(provider, "_detect_branch_name", return_value="develop"):
            temp_dir, branch_name = provider._clone_repository(
                url, github_username="user", personal_access_token="token123"
            )
        expected_url = "https://user:token123@github.com/user/repo.git"
        mock_clone.assert_called_with(
            expected_url,
            "/tmp/fake-dir",
            depth=1,
            config=mock.ANY,
            pool_manager=mock.ANY,
        )
        assert temp_dir == "/tmp/fake-dir"
        assert branch_name == "develop"

    @mock.patch("prowler.providers.iac.iac_provider.porcelain.clone")
    @mock.patch("tempfile.mkdtemp", return_value="/tmp/fake-dir")
    def test_clone_repository_with_oauth(self, _mock_mkdtemp, mock_clone):
        provider = IacProvider()
        url = "https://github.com/user/repo.git"
        with mock.patch.object(provider, "_detect_branch_name", return_value="master"):
            temp_dir, branch_name = provider._clone_repository(
                url, oauth_app_token="oauth456"
            )
        expected_url = "https://oauth2:oauth456@github.com/user/repo.git"
        mock_clone.assert_called_with(
            expected_url,
            "/tmp/fake-dir",
            depth=1,
            config=mock.ANY,
            pool_manager=mock.ANY,
        )
        assert temp_dir == "/tmp/fake-dir"
        assert branch_name == "master"

    @mock.patch("prowler.providers.iac.iac_provider.porcelain.clone")
    @mock.patch("tempfile.mkdtemp", return_value="/tmp/fake-dir")
    def test_clone_repository_validates_redirect_destinations(
        self, _mock_mkdtemp, mock_clone
    ):
        """dulwich adopts a redirect's destination, so the clone needs the guarded manager."""
        provider = IacProvider()
        with (
            mock.patch.object(provider, "_detect_branch_name", return_value="main"),
            mock.patch(
                "prowler.providers.iac.iac_provider.guarded_pool_manager",
                return_value="the-guarded-manager",
            ),
        ):
            provider._clone_repository("https://github.com/user/repo.git")

        assert mock_clone.call_args.kwargs["pool_manager"] == "the-guarded-manager"

    @mock.patch("prowler.providers.iac.iac_provider.porcelain.clone")
    @mock.patch("tempfile.mkdtemp", return_value="/tmp/fake-dir")
    def test_clone_repository_failure_raises(self, _mock_mkdtemp, mock_clone):
        """A failed clone must raise a typed error instead of returning `None`.

        `_clone_repository` is annotated `-> tuple[str, str]` and `__init__`
        unpacks the result directly, so falling through the error handler used
        to raise `TypeError: cannot unpack non-sequence NoneType`. Raising
        `IacRepositoryCloneError` lets the CLI exit with the logged message and
        lets the API report the failure as a normal task error instead of a
        `SystemExit` escaping the Celery worker.
        """
        mock_clone.side_effect = Exception(
            "https://x-access-token:SENTINEL_TOKEN@github.com/user/repo.git refused"
        )

        with pytest.raises(IacRepositoryCloneError) as exc_info:
            IacProvider(scan_repository_url="https://github.com/user/repo.git")

        # ProwlerException formats original_exception into its str(), and the API
        # returns that, so the authenticated URL must not reach it
        assert "SENTINEL_TOKEN" not in str(exc_info.value)
        assert "Exception" in str(exc_info.value)
        assert exc_info.value.code == 21000
        assert isinstance(exc_info.value.__cause__, Exception)

    def test_detect_branch_name_main(self):
        """Test detecting 'main' branch from .git/HEAD"""
        provider = IacProvider()
        with tempfile.TemporaryDirectory() as temp_dir:
            # Create a mock .git/HEAD file with main branch
            git_dir = os.path.join(temp_dir, ".git")
            os.makedirs(git_dir)
            head_file = os.path.join(git_dir, "HEAD")
            with open(head_file, "w") as f:
                f.write("ref: refs/heads/main\n")

            branch_name = provider._detect_branch_name(temp_dir)
            assert branch_name == "main"

    def test_detect_branch_name_custom_branch(self):
        """Test detecting custom branch like 'develop' from .git/HEAD"""
        provider = IacProvider()
        with tempfile.TemporaryDirectory() as temp_dir:
            # Create a mock .git/HEAD file with develop branch
            git_dir = os.path.join(temp_dir, ".git")
            os.makedirs(git_dir)
            head_file = os.path.join(git_dir, "HEAD")
            with open(head_file, "w") as f:
                f.write("ref: refs/heads/develop\n")

            branch_name = provider._detect_branch_name(temp_dir)
            assert branch_name == "develop"

    def test_detect_branch_name_fallback(self):
        """Test fallback to 'main' when .git/HEAD doesn't exist"""
        provider = IacProvider()
        with tempfile.TemporaryDirectory() as temp_dir:
            # Don't create .git/HEAD file
            branch_name = provider._detect_branch_name(temp_dir)
            assert branch_name == "main"

    def test_detect_branch_name_error_handling(self):
        """Test error handling returns 'main' as fallback"""
        provider = IacProvider()
        # Pass a non-existent directory
        branch_name = provider._detect_branch_name("/non/existent/path")
        assert branch_name == "main"

    def test_test_connection_rejects_loopback_url(self):
        with patch("prowler.providers.iac.iac_provider.ls_remote") as mock_ls_remote:
            connection = IacProvider.test_connection(
                scan_repository_url="https://127.0.0.1/user/repo.git"
            )

        assert connection.is_connected is False
        assert connection.error == "Repository URL is not an allowed destination."
        mock_ls_remote.assert_not_called()

    def test_test_connection_rejects_private_range_url(self):
        with patch("prowler.providers.iac.iac_provider.ls_remote") as mock_ls_remote:
            connection = IacProvider.test_connection(
                scan_repository_url="https://10.0.0.1/user/repo.git"
            )

        assert connection.is_connected is False
        assert connection.error == "Repository URL is not an allowed destination."
        mock_ls_remote.assert_not_called()

    def test_clone_repository_rejects_a_non_public_url(self):
        provider = IacProvider.__new__(IacProvider)
        with patch("prowler.providers.iac.iac_provider.porcelain.clone") as mock_clone:
            with pytest.raises(OutboundURLNotAllowedError):
                provider._clone_repository("https://169.254.169.254/org/repo.git")

        mock_clone.assert_not_called()

    def test_clone_repository_rejects_shared_address_space(self):
        provider = IacProvider.__new__(IacProvider)
        with patch("prowler.providers.iac.iac_provider.porcelain.clone") as mock_clone:
            with pytest.raises(OutboundURLNotAllowedError):
                provider._clone_repository("https://100.100.100.200/org/repo.git")

        mock_clone.assert_not_called()

    def test_clone_repository_does_not_validate_the_token_bearing_url(self):
        provider = IacProvider.__new__(IacProvider)
        with patch(
            "prowler.lib.network.ssrf.socket.getaddrinfo",
            return_value=[(None, None, None, None, ("140.82.121.4", 0))],
        ) as mock_getaddrinfo:
            with patch("prowler.providers.iac.iac_provider.porcelain.clone"):
                with patch.object(
                    IacProvider, "_detect_branch_name", return_value="main"
                ):
                    provider._clone_repository(
                        "https://github.com/org/repo.git",
                        github_username="user",
                        personal_access_token="token",
                    )

        assert mock_getaddrinfo.call_args[0][0] == "github.com"

    def test_test_connection_allows_public_url(self):
        with (
            patch(
                "prowler.lib.network.ssrf.socket.getaddrinfo",
                return_value=[(None, None, None, None, ("140.82.121.4", 0))],
            ) as mock_getaddrinfo,
            patch("prowler.providers.iac.iac_provider.ls_remote") as mock_ls_remote,
        ):
            connection = IacProvider.test_connection(
                scan_repository_url="https://github.com/user/repo.git"
            )

        assert connection.is_connected is True
        mock_getaddrinfo.assert_called_once_with("github.com", None)
        mock_ls_remote.assert_called_once_with("https://github.com/user/repo.git")

    def test_test_connection_does_not_echo_raw_error(self):
        with (
            patch(
                "prowler.lib.network.ssrf.socket.getaddrinfo",
                return_value=[(None, None, None, None, ("140.82.121.4", 0))],
            ),
            patch(
                "prowler.providers.iac.iac_provider.ls_remote",
                side_effect=Exception(
                    "https://x-access-token:SENTINEL_TOKEN@github.com/user/repo.git refused"
                ),
            ),
        ):
            connection = IacProvider.test_connection(
                scan_repository_url="https://github.com/user/repo.git",
                oauth_app_token="SENTINEL_TOKEN",
            )

        assert connection.is_connected is False
        assert connection.error == "Failed to connect to repository."
        assert "SENTINEL_TOKEN" not in connection.error