diff --git a/src/scp/scp_utils.py b/src/scp/scp_utils.py index 824e980..b94a55b 100644 --- a/src/scp/scp_utils.py +++ b/src/scp/scp_utils.py @@ -27,6 +27,16 @@ _pkg_dir = Path(__file__).resolve().parent +def _new_quarantine_id(qdir: Path) -> str: + for _ in range(10): + qid = uuid.uuid4().hex + content_path = qdir / f"{qid}.txt" + meta_path = qdir / f"{qid}.json" + if not content_path.exists() and not meta_path.exists(): + return qid + raise ValueError("unable to allocate unique quarantine_id") + + def _quarantine_dir() -> Path: env = os.environ.get("SCP_QUARANTINE_DIR") if env: @@ -163,7 +173,7 @@ def quarantine( else: raise ValueError(f"unsupported quarantine layout: {layout!r}") qdir.mkdir(parents=True, exist_ok=True) - qid = str(uuid.uuid4())[:8] + qid = _new_quarantine_id(qdir) meta = {"quarantine_id": qid, "reason": reason, "source": source} meta_json = json.dumps(meta, indent=2) content_bytes = len(content.encode("utf-8", errors="replace")) diff --git a/tests/test_security_regressions.py b/tests/test_security_regressions.py index bbbc897..b61eb84 100644 --- a/tests/test_security_regressions.py +++ b/tests/test_security_regressions.py @@ -4,6 +4,7 @@ import os import time +import uuid import pytest @@ -119,6 +120,26 @@ def test_quarantine_impossible_write_does_not_evict_existing_entries(tmp_path, m assert old_json.is_file() +def test_quarantine_uses_full_uuid_to_avoid_prefix_collision(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("SCP_QUARANTINE_DIR", str(tmp_path)) + ids = iter( + [ + uuid.UUID("aaaaaaaa-0000-0000-0000-000000000001"), + uuid.UUID("aaaaaaaa-0000-0000-0000-000000000002"), + ] + ) + monkeypatch.setattr(scp_utils.uuid, "uuid4", lambda: next(ids)) + + first = scp_utils.quarantine("first quarantine evidence", reason="r1", source="s1") + second = scp_utils.quarantine("second quarantine evidence", reason="r2", source="s2") + + assert first["quarantine_id"] != second["quarantine_id"] + first_path = tmp_path / f"{first['quarantine_id']}.txt" + second_path = tmp_path / f"{second['quarantine_id']}.txt" + assert first_path.read_text(encoding="utf-8") == "first quarantine evidence" + assert second_path.read_text(encoding="utf-8") == "second quarantine evidence" + + def test_registry_fetch_layout_counts_toward_total_quota(tmp_path, monkeypatch) -> None: """AppSec: registry_fetch/ writes must count toward SCP_QUARANTINE_MAX_TOTAL_BYTES.""" from scp import quarantine_limits