ITWorx_MCP_Hub-Public/tests/test_wp160_security.py

from __future__ import annotations

import json
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from uuid import uuid4

from packages.connector_kit.gitea import GiteaSettings
from packages.connector_kit.knowledge import (
    DocumentNotFound,
    InMemoryKnowledgeIndex,
    PathEscapeError,
    RegisteredSource,
)
from packages.connector_kit.redaction import safe_content
from packages.hub_contracts.dataplane import AliasRecord
from scripts.generate_sbom import generate
from scripts.security_scan import (
    _BEARER,
    _tracked_text_files,
    scan_repository,
    scan_secret_patterns,
)
from services.control_api.catalog.models import CatalogState
from services.control_api.catalog.planner import build_plan
from services.control_api.contextforge.models import GatewayDesiredState, GatewaySnapshot


class SecurityPlanTests(unittest.TestCase):
    def test_bearer_secret_scan_does_not_cross_configuration_lines(self) -> None:
        self.assertIsNone(_BEARER.search("HUB_CONTEXTFORGE_AUTH_SCHEME: Bearer\nNEXT_KEY=value"))
        self.assertIsNotNone(_BEARER.search("Authorization: Bearer abcdefghijklmnopqrstuvwxyz1234"))

    def test_static_security_scan_proves_compose_invariants_and_no_live_secret(self) -> None:
        report = scan_repository()
        self.assertEqual(report["status"], "pass")
        self.assertEqual(report["compose"]["violations"], [])
        self.assertEqual(report["secrets"]["findings"], [])
        self.assertTrue(report["invariants"]["denyByDefault"])

    def test_secret_scan_ignores_untracked_scratch_checkout_files(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            repository = Path(directory)
            git = shutil.which("git")
            self.assertIsNotNone(git)
            assert git is not None
            subprocess.run([git, "init", "--quiet"], cwd=repository, check=True)  # noqa: S603
            tracked = repository / "tracked.md"
            tracked.write_text("ordinary tracked documentation\n", encoding="utf-8")
            subprocess.run([git, "add", "tracked.md"], cwd=repository, check=True)  # noqa: S603
            tracked_secret = repository / "tracked-secret.md"
            tracked_secret.write_text(
                'password = "tracked-sensitive-value-9283746501"\n', encoding="utf-8"
            )
            subprocess.run([git, "add", "tracked-secret.md"], cwd=repository, check=True)  # noqa: S603
            scratch = repository / ".tmp" / "source-checkout"
            scratch.mkdir(parents=True)
            (scratch / "ignored.md").write_text(
                'password = "untracked-sensitive-value-9283746501"\n', encoding="utf-8"
            )

            with patch("scripts.security_scan.ROOT", repository):
                self.assertCountEqual(_tracked_text_files(), [tracked, tracked_secret])
                findings = scan_secret_patterns()["findings"]
                self.assertEqual(len(findings), 1)
                self.assertEqual(findings[0]["path"], "tracked-secret.md")

    def test_git_enumeration_failure_fails_closed(self) -> None:
        error = subprocess.CalledProcessError(2, ["git", "ls-files", "-z"])
        with patch("scripts.security_scan.subprocess.run", side_effect=error):
            with self.assertRaises(subprocess.CalledProcessError):
                _tracked_text_files()

    def test_missing_git_fails_closed(self) -> None:
        with patch("scripts.security_scan.shutil.which", return_value=None):
            with self.assertRaises(FileNotFoundError):
                _tracked_text_files()

    def test_sbom_is_deterministic_and_contains_locked_dependencies_and_images(self) -> None:
        report = generate()
        self.assertEqual(report["bomFormat"], "CycloneDX")
        self.assertEqual(report["metadata"]["component"]["name"], "itworx-mcp-hub")
        self.assertTrue(report["metadata"]["sourceCommit"])
        components = report["components"]
        self.assertTrue(any(item["name"] == "fastapi" for item in components))
        self.assertTrue(
            any(
                item["type"] == "container" and "@sha256:" in item["version"] for item in components
            )
        )

    def test_prompt_injection_is_data_and_secret_content_is_rejected(self) -> None:
        fixture = json.loads(
            (
                Path(__file__).resolve().parents[1]
                / "tests/fixtures/security/wp160-untrusted-inputs.json"
            ).read_text(encoding="utf-8")
        )
        content = fixture["promptInjection"]
        self.assertEqual(safe_content(content), content)
        self.assertIsNone(safe_content(fixture["secretLike"]))

    def test_registered_paths_and_document_uris_cannot_escape(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            source = RegisteredSource(
                source_id=uuid4(),
                project_id=uuid4(),
                project_slug="ludarium",
                source_ref="docs",
                source_type="documentation",
                root=Path(directory),
            )
            with self.assertRaises(PathEscapeError):
                source.canonical_path("../secrets")
        with self.assertRaises(DocumentNotFound):
            InMemoryKnowledgeIndex().get_document(uuid4(), "itworx-docs://ludarium/docs/%2e%2e")

    def test_ssrf_and_token_substitution_boundaries_are_rejected(self) -> None:
        with self.assertRaises(ValueError):
            GiteaSettings(
                base_url="https://169.254.169.254", token_reference="docker-" + "secret://gitea"
            )
        with self.assertRaises(ValueError):
            GiteaSettings(
                base_url="https://localhost", token_reference="docker-" + "secret://gitea"
            )
        with self.assertRaises(ValueError):
            AliasRecord(
                project_slug="ludarium",
                environment="prod",
                alias="/mcp/ludarium/prod",
                upstream_url="https://127.0.0.1:4444/mcp",
            )

    def test_gateway_drift_is_reported_and_unknown_gateway_is_not_deleted(self) -> None:
        desired = GatewayDesiredState(slug="knowledge", url="https://knowledge.example/mcp")
        unknown = GatewaySnapshot(
            slug="operator-owned",
            url="https://operator.example/mcp",
            description="outside catalog",
            transport="STREAMABLEHTTP",
            enabled=True,
            status="ready",
        )
        state = CatalogState(
            documents=(),
            gateways=(desired,),
            desired_hash="0" * 64,
            source_files=("catalog/project.yaml",),
        )
        plan = build_plan(state, (unknown,), set())
        self.assertEqual(plan.actions[0].action, "create")
        self.assertEqual(plan.drift[0].kind, "unknown_gateway")
        self.assertNotIn("delete", {action.action for action in plan.actions})

    def test_json_scan_output_is_serializable_without_credentials(self) -> None:
        payload = json.dumps(scan_repository(), sort_keys=True)
        self.assertNotIn("BEGIN PRIVATE KEY", payload)
        self.assertNotIn("Authorization: Bearer", payload)


if __name__ == "__main__":
    unittest.main()