122 lines
5.0 KiB
Python
122 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
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, scan_repository
|
|
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_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()
|