mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 01:48:17 +00:00
feat(access-control-prover): migrate reviewed entry packs
This commit is contained in:
Marius Ciepluch
@@ -5,12 +5,21 @@ import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from openultrasast.core.factpack import FactPackError, admit_pack_file, verify_fact_pack # noqa: E402
|
||||
from openultrasast.mapping.installed_fact_authority import ( # noqa: E402
|
||||
InstalledAuthorityWireError,
|
||||
decode_fact_pack_bytes,
|
||||
)
|
||||
|
||||
REPO = Path(__file__).resolve().parents[1]
|
||||
PACK = REPO / "src/openultrasast/ruleset/entries/express.json"
|
||||
ENTRIES = REPO / "src/openultrasast/ruleset/entries"
|
||||
LEGACY_DIGEST = "e8e149309abd78f24206c58fde7f73ae2f60364d8b8fee22ecaee8342798d8ce"
|
||||
PACK = ENTRIES / "artifacts" / "sha256" / LEGACY_DIGEST
|
||||
PROJECTION = ENTRIES / "express.json"
|
||||
|
||||
|
||||
def _review_digest() -> str:
|
||||
@@ -19,6 +28,9 @@ def _review_digest() -> str:
|
||||
|
||||
|
||||
def test_express_pack_admitted_with_verifiable_review() -> None:
|
||||
content = PACK.read_bytes()
|
||||
assert len(content) == 963
|
||||
assert hashlib.sha256(content).hexdigest() == LEGACY_DIGEST
|
||||
pack = admit_pack_file(PACK, review_digest=_review_digest())
|
||||
assert pack.family == "access_control"
|
||||
assert len(pack.facts) == 3
|
||||
@@ -34,19 +46,20 @@ def test_pack_refused_without_matching_review() -> None:
|
||||
pass
|
||||
|
||||
|
||||
def test_tampered_pack_file_refused() -> None:
|
||||
def test_tampered_pack_file_refused(tmp_path: Path) -> None:
|
||||
payload = json.loads(PACK.read_text(encoding="utf-8"))
|
||||
payload["facts"].append({"id": "entry:express:evil", "pattern": "eval("})
|
||||
tampered = PACK.parent / "tampered.json"
|
||||
tampered = tmp_path / "tampered.json"
|
||||
tampered.write_text(json.dumps(payload), encoding="utf-8")
|
||||
try:
|
||||
pack = admit_pack_file(tampered, review_digest=_review_digest())
|
||||
# review attestation still verifies, but content no longer matches the stamped digest
|
||||
original = admit_pack_file(PACK, review_digest=_review_digest())
|
||||
try:
|
||||
verify_fact_pack(original, pack.facts)
|
||||
raise AssertionError("tampered content must fail verification")
|
||||
except FactPackError:
|
||||
pass
|
||||
finally:
|
||||
tampered.unlink()
|
||||
pack = admit_pack_file(tampered, review_digest=_review_digest())
|
||||
# review attestation still verifies, but content no longer matches the stamped digest
|
||||
original = admit_pack_file(PACK, review_digest=_review_digest())
|
||||
with pytest.raises(FactPackError):
|
||||
verify_fact_pack(original, pack.facts)
|
||||
|
||||
|
||||
def test_compatibility_projection_is_neither_a_legacy_nor_v2_fact_pack() -> None:
|
||||
with pytest.raises(FactPackError):
|
||||
admit_pack_file(PROJECTION, review_digest=_review_digest())
|
||||
with pytest.raises(InstalledAuthorityWireError):
|
||||
decode_fact_pack_bytes(PROJECTION.read_bytes())
|
||||
|
||||
+271
-2
@@ -1,16 +1,39 @@
|
||||
"""Entry-pack wiring tests for detection-core 8.2 part 2 (Req 10, 3)."""
|
||||
"""Installed reviewed entry-pack migration and legacy recognition wiring."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from openultrasast.mapping.entry_packs import load_entry_pack_patterns # noqa: E402
|
||||
from openultrasast.core import canonical, provenance, semantics # noqa: E402
|
||||
from openultrasast.core.factpack import ( # noqa: E402
|
||||
StructuralRequirementKind,
|
||||
validate_fact_pack,
|
||||
)
|
||||
from openultrasast.mapping import entry_packs as entry_pack_module # noqa: E402
|
||||
from openultrasast.mapping.entry_packs import ( # noqa: E402
|
||||
load_entry_pack_patterns,
|
||||
load_installed_entry_fact_authority,
|
||||
)
|
||||
from openultrasast.mapping.fact_authority import ( # noqa: E402
|
||||
EntryFactAuthorityContext,
|
||||
EntryFactAuthorityError,
|
||||
)
|
||||
from openultrasast.mapping.installed_fact_authority import ( # noqa: E402
|
||||
InstalledAuthorityWireError,
|
||||
decode_fact_pack_bytes,
|
||||
)
|
||||
from openultrasast.mapping.legacy import _js_entry_points, analyze_entry_points, entry_framework # noqa: E402
|
||||
from openultrasast.preprocess import FileTarget, detect_language # noqa: E402
|
||||
|
||||
REPO = Path(__file__).resolve().parents[1]
|
||||
ENTRIES = REPO / "src" / "openultrasast" / "ruleset" / "entries"
|
||||
|
||||
TS_SOURCE = """import express from "express";
|
||||
const app = express();
|
||||
@@ -18,6 +41,61 @@ app.all("/admin", adminHandler);
|
||||
function adminHandler(req, res) { res.send("ok"); }
|
||||
"""
|
||||
|
||||
_LEGACY_ROWS = (
|
||||
(
|
||||
"entry:express:app-route",
|
||||
True,
|
||||
"route",
|
||||
r"app.(get|post|put|delete|patch|all|use)\s*\(",
|
||||
),
|
||||
(
|
||||
"entry:express:router-route",
|
||||
True,
|
||||
"route",
|
||||
r"router.(get|post|put|delete|patch|all|use)\s*\(",
|
||||
),
|
||||
("entry:express:express-app", False, "app", r"express\s*\(\s*\)"),
|
||||
)
|
||||
|
||||
|
||||
class _UnusedCustody:
|
||||
def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog:
|
||||
raise AssertionError(f"authority authentication read target custody: {population_digest}")
|
||||
|
||||
|
||||
class _InstalledCustody:
|
||||
def __init__(self, logs: dict[str, provenance.PopulationCustodyLog]) -> None:
|
||||
self._logs = logs
|
||||
|
||||
def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog:
|
||||
return self._logs[population_digest]
|
||||
|
||||
|
||||
def _custody_for_installed_lineages(
|
||||
lineages: tuple[provenance.LineageManifest, ...],
|
||||
policy: provenance.PopulationCustodyPolicy,
|
||||
) -> _InstalledCustody:
|
||||
custodian = policy.reviewed_custodian_digests[0]
|
||||
logs: dict[str, provenance.PopulationCustodyLog] = {}
|
||||
for lineage in lineages:
|
||||
for use in lineage.uses:
|
||||
prior = logs.setdefault(
|
||||
use.source_population_digest,
|
||||
provenance.empty_population_custody_log(use.source_population_digest, custodian),
|
||||
)
|
||||
request = provenance.PopulationUseRequest(
|
||||
use.source_population_digest,
|
||||
lineage.arm_id,
|
||||
use.purpose,
|
||||
use.labels_accessed,
|
||||
use.influenced_artifact_digest,
|
||||
prior.log_digest,
|
||||
)
|
||||
receipt = provenance.record_population_use(request, prior, custodian, policy)
|
||||
assert receipt.receipt_digest == use.population_use_receipt_digest
|
||||
logs[use.source_population_digest] = provenance.append_population_use(prior, receipt, policy)
|
||||
return _InstalledCustody(logs)
|
||||
|
||||
|
||||
def _target(root: Path, name: str, text: str) -> FileTarget:
|
||||
path = root / name
|
||||
@@ -26,11 +104,202 @@ def _target(root: Path, name: str, text: str) -> FileTarget:
|
||||
return FileTarget(name, str(path), detect_language(path), 5, [], False, [], [])
|
||||
|
||||
|
||||
def _stage_entries(root: Path) -> Path:
|
||||
staged = root / "ruleset" / "entries"
|
||||
shutil.copytree(ENTRIES, staged)
|
||||
return staged
|
||||
|
||||
|
||||
def test_pack_patterns_loaded_for_typescript() -> None:
|
||||
patterns = load_entry_pack_patterns()
|
||||
assert patterns.get("typescript") and any(p.search("app.all(") for p in patterns["typescript"])
|
||||
|
||||
|
||||
def test_installed_v2_packs_authenticate_and_validate_from_a_proven_legacy_input() -> None:
|
||||
context = load_installed_entry_fact_authority(_UnusedCustody())
|
||||
bundle = context.bundle
|
||||
legacy_reference = bundle.packs[0].review_evidence[0].content
|
||||
legacy_artifact = REPO / "src" / "openultrasast" / "ruleset" / "entries" / "artifacts" / "sha256" / legacy_reference.content_digest
|
||||
|
||||
legacy_bytes = legacy_artifact.read_bytes()
|
||||
assert legacy_artifact.stat().st_size == 963
|
||||
assert len(legacy_bytes) == 963
|
||||
assert canonical.bytes_digest(legacy_bytes) == legacy_reference.content_digest
|
||||
|
||||
assert tuple(pack.language for pack in bundle.packs) == ("javascript", "typescript")
|
||||
custody = _custody_for_installed_lineages(bundle.lineages, bundle.custody_policy)
|
||||
for pack in bundle.packs:
|
||||
validate_fact_pack(
|
||||
pack,
|
||||
bundle.contract,
|
||||
None,
|
||||
bundle.lineages,
|
||||
custody,
|
||||
bundle.custody_policy,
|
||||
bundle.review_policy,
|
||||
context.resolver,
|
||||
)
|
||||
|
||||
|
||||
def test_reviewed_pack_mapping_is_exact_and_language_scoped() -> None:
|
||||
bundle = load_installed_entry_fact_authority(_UnusedCustody()).bundle
|
||||
methods = ("get", "post", "put", "delete", "patch", "all", "use")
|
||||
|
||||
for pack in bundle.packs:
|
||||
facts = {fact.fact_id: fact for fact in pack.facts}
|
||||
assert set(facts) == {
|
||||
*(f"entry:express:app-route:{method}" for method in methods),
|
||||
*(f"entry:express:router-route:{method}" for method in methods),
|
||||
"entry:express:express-app",
|
||||
}
|
||||
for receiver_name, receiver_type in (("app", "express_application"), ("router", "express_router")):
|
||||
for method in methods:
|
||||
fact = facts[f"entry:express:{receiver_name}-route:{method}"]
|
||||
assert fact.role == "entry_registration"
|
||||
assert tuple((item.requirement_id, item.kind, item.field, item.expected) for item in fact.structural_requirements) == (
|
||||
("qualified_target", StructuralRequirementKind.QUALIFIED_NAME, "qualified_target", f"{receiver_name}.{method}"),
|
||||
("call_shape", StructuralRequirementKind.AST_SHAPE, "call_shape", "member_call"),
|
||||
("receiver", StructuralRequirementKind.RECEIVER_TYPE, "receiver", receiver_type),
|
||||
("arguments", StructuralRequirementKind.ARGUMENT_SHAPE, "arguments", "route_path_and_handler"),
|
||||
)
|
||||
assert tuple(
|
||||
(effect.kind, effect.role, effect.source_role, effect.target_role, effect.conditions)
|
||||
for effect in fact.relation_effects
|
||||
) == ((semantics.RelationKind.REGISTRATION, "entry_registration", "framework_instance", "entry_handler", ()),)
|
||||
app = facts["entry:express:express-app"]
|
||||
assert app.role == "framework_instance"
|
||||
assert tuple((item.requirement_id, item.kind, item.field, item.expected) for item in app.structural_requirements) == (
|
||||
("qualified_target", StructuralRequirementKind.QUALIFIED_NAME, "qualified_target", "express"),
|
||||
("call_shape", StructuralRequirementKind.AST_SHAPE, "call_shape", "factory_call"),
|
||||
("binding", StructuralRequirementKind.ARGUMENT_SHAPE, "binding", "application_instance_binding"),
|
||||
)
|
||||
assert tuple(
|
||||
(effect.kind, effect.role, effect.source_role, effect.target_role, effect.conditions) for effect in app.relation_effects
|
||||
) == ((semantics.RelationKind.REGISTRATION, "framework_instance", "framework_factory", "framework_instance", ()),)
|
||||
|
||||
|
||||
def test_projection_preserves_legacy_rows_but_is_not_a_fact_pack() -> None:
|
||||
projection_path = REPO / "src" / "openultrasast" / "ruleset" / "entries" / "express.json"
|
||||
content = projection_path.read_bytes()
|
||||
raw = json.loads(content)
|
||||
|
||||
assert content == json.dumps(raw, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
|
||||
assert raw["schema_version"] == "openultrasast.entry-recognition-projection.v1"
|
||||
assert (
|
||||
tuple(
|
||||
(
|
||||
row["legacy_fact_id"],
|
||||
row["defines_handler"],
|
||||
row["handler_kind"],
|
||||
row["pattern"],
|
||||
)
|
||||
for row in raw["facts"]
|
||||
)
|
||||
== _LEGACY_ROWS
|
||||
)
|
||||
assert all(set(row) == {"legacy_fact_id", "languages", "defines_handler", "handler_kind", "pattern"} for row in raw["facts"])
|
||||
with pytest.raises(InstalledAuthorityWireError):
|
||||
decode_fact_pack_bytes(content)
|
||||
|
||||
|
||||
def test_repeated_pattern_load_reuses_fully_authenticated_state(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_stage_entries(tmp_path)
|
||||
entry_pack_module._cache.clear()
|
||||
authenticates = 0
|
||||
real_load = entry_pack_module.load_installed_entry_fact_authority
|
||||
|
||||
def counted_load(
|
||||
custody: provenance.PopulationCustodyPort,
|
||||
repo_root: Path | None = None,
|
||||
) -> EntryFactAuthorityContext:
|
||||
nonlocal authenticates
|
||||
authenticates += 1
|
||||
return real_load(custody, repo_root)
|
||||
|
||||
monkeypatch.setattr(entry_pack_module, "load_installed_entry_fact_authority", counted_load)
|
||||
|
||||
first = load_entry_pack_patterns(tmp_path)
|
||||
second = load_entry_pack_patterns(tmp_path)
|
||||
|
||||
assert authenticates == 1
|
||||
assert first is not second
|
||||
assert tuple(pattern.pattern for pattern in first["javascript"]) == tuple(pattern.pattern for pattern in second["javascript"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"relative_path",
|
||||
(
|
||||
"bootstrap.json",
|
||||
"identities/generator.json",
|
||||
"authority.json",
|
||||
"express.javascript.v2.json",
|
||||
"express.json",
|
||||
"artifacts/sha256/529d18ba2b351a8de985a9cf416261d717ef55a4b7a9d9d65bce53b97d59ab35",
|
||||
),
|
||||
)
|
||||
def test_pattern_cache_invalidates_for_every_authority_state_class(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
relative_path: str,
|
||||
) -> None:
|
||||
entries = _stage_entries(tmp_path)
|
||||
entry_pack_module._cache.clear()
|
||||
authenticates = 0
|
||||
real_load = entry_pack_module.load_installed_entry_fact_authority
|
||||
|
||||
def counted_load(
|
||||
custody: provenance.PopulationCustodyPort,
|
||||
repo_root: Path | None = None,
|
||||
) -> EntryFactAuthorityContext:
|
||||
nonlocal authenticates
|
||||
authenticates += 1
|
||||
return real_load(custody, repo_root)
|
||||
|
||||
monkeypatch.setattr(entry_pack_module, "load_installed_entry_fact_authority", counted_load)
|
||||
load_entry_pack_patterns(tmp_path)
|
||||
changed = entries / relative_path
|
||||
stat = changed.stat()
|
||||
os.utime(changed, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000))
|
||||
|
||||
load_entry_pack_patterns(tmp_path)
|
||||
|
||||
assert authenticates == 2
|
||||
|
||||
|
||||
def test_pattern_cache_fingerprints_content_even_when_size_and_mtime_match(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
entries = _stage_entries(tmp_path)
|
||||
entry_pack_module._cache.clear()
|
||||
authenticates = 0
|
||||
real_load = entry_pack_module.load_installed_entry_fact_authority
|
||||
|
||||
def counted_load(
|
||||
custody: provenance.PopulationCustodyPort,
|
||||
repo_root: Path | None = None,
|
||||
) -> EntryFactAuthorityContext:
|
||||
nonlocal authenticates
|
||||
authenticates += 1
|
||||
return real_load(custody, repo_root)
|
||||
|
||||
monkeypatch.setattr(entry_pack_module, "load_installed_entry_fact_authority", counted_load)
|
||||
load_entry_pack_patterns(tmp_path)
|
||||
artifact = entries / "artifacts" / "sha256" / "529d18ba2b351a8de985a9cf416261d717ef55a4b7a9d9d65bce53b97d59ab35"
|
||||
content = artifact.read_bytes()
|
||||
stat = artifact.stat()
|
||||
artifact.write_bytes(bytes((content[0] ^ 1,)) + content[1:])
|
||||
os.utime(artifact, ns=(stat.st_atime_ns, stat.st_mtime_ns))
|
||||
|
||||
with pytest.raises(EntryFactAuthorityError):
|
||||
load_entry_pack_patterns(tmp_path)
|
||||
|
||||
assert authenticates == 2
|
||||
|
||||
|
||||
def test_pack_patterns_resolve_from_package_not_target_repo() -> None:
|
||||
# default call reads the installed package's ruleset, independent of any analyzed repo
|
||||
assert load_entry_pack_patterns().get("typescript")
|
||||
|
||||
Reference in New Issue
Block a user