fix(assessment): constrain installed record reads
This commit is contained in:
@@ -11,6 +11,7 @@ import sys
|
|||||||
import tempfile
|
import tempfile
|
||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
import unittest
|
import unittest
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
from cryptography.hazmat.primitives import serialization
|
from cryptography.hazmat.primitives import serialization
|
||||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||||||
@@ -217,11 +218,12 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
|
|||||||
root = Path(temp_dir)
|
root = Path(temp_dir)
|
||||||
distribution, payload_file = create_distribution(root)
|
distribution, payload_file = create_distribution(root)
|
||||||
|
|
||||||
evidence = collect_installed_composition(
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
||||||
assessment=self.assessment,
|
evidence = collect_installed_composition(
|
||||||
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
assessment=self.assessment,
|
||||||
distributions=[distribution],
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
||||||
)
|
distributions=[distribution],
|
||||||
|
)
|
||||||
|
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
(), validate_payload(payload=evidence, schema=self.installed_schema)
|
(), validate_payload(payload=evidence, schema=self.installed_schema)
|
||||||
@@ -234,11 +236,12 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
|
|||||||
self.assertNotIn("file://", encoded)
|
self.assertNotIn("file://", encoded)
|
||||||
|
|
||||||
payload_file.write_text("tampered\n", encoding="utf-8")
|
payload_file.write_text("tampered\n", encoding="utf-8")
|
||||||
tampered = collect_installed_composition(
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
||||||
assessment=self.assessment,
|
tampered = collect_installed_composition(
|
||||||
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
assessment=self.assessment,
|
||||||
distributions=[distribution],
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
||||||
)
|
distributions=[distribution],
|
||||||
|
)
|
||||||
|
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
"mismatch", tampered["artifacts"][0]["record_integrity"]["status"]
|
"mismatch", tampered["artifacts"][0]["record_integrity"]["status"]
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ import hashlib
|
|||||||
import hmac
|
import hmac
|
||||||
from importlib import metadata
|
from importlib import metadata
|
||||||
import json
|
import json
|
||||||
from pathlib import Path
|
from pathlib import Path, PurePosixPath
|
||||||
import re
|
import re
|
||||||
import sys
|
import sys
|
||||||
from typing import Any, Iterable, Mapping, Sequence
|
from typing import Any, Iterable, Mapping, Sequence
|
||||||
@@ -1080,6 +1080,12 @@ def _record_integrity(
|
|||||||
installation_root = Path(sys.prefix).resolve()
|
installation_root = Path(sys.prefix).resolve()
|
||||||
except OSError:
|
except OSError:
|
||||||
return {"status": "unavailable", **counts}
|
return {"status": "unavailable", **counts}
|
||||||
|
if not _trusted_distribution_root(distribution_root):
|
||||||
|
return {"status": "unavailable", **counts}
|
||||||
|
script_roots = (
|
||||||
|
(installation_root / "bin").resolve(),
|
||||||
|
(installation_root / "Scripts").resolve(),
|
||||||
|
)
|
||||||
total_hashed_bytes = 0
|
total_hashed_bytes = 0
|
||||||
limit_exceeded = False
|
limit_exceeded = False
|
||||||
for package_path in files:
|
for package_path in files:
|
||||||
@@ -1098,10 +1104,18 @@ def _record_integrity(
|
|||||||
except OSError:
|
except OSError:
|
||||||
counts["missing_file_count"] += 1
|
counts["missing_file_count"] += 1
|
||||||
continue
|
continue
|
||||||
if not (
|
record_path = PurePosixPath(str(package_path).replace("\\", "/"))
|
||||||
_is_relative_to(resolved_path, distribution_root)
|
has_parent_traversal = ".." in record_path.parts
|
||||||
or _is_relative_to(resolved_path, installation_root)
|
path_allowed = (
|
||||||
):
|
not record_path.is_absolute()
|
||||||
|
and not has_parent_traversal
|
||||||
|
and _is_relative_to(resolved_path, distribution_root)
|
||||||
|
) or (
|
||||||
|
not record_path.is_absolute()
|
||||||
|
and has_parent_traversal
|
||||||
|
and any(_is_relative_to(resolved_path, root) for root in script_roots)
|
||||||
|
)
|
||||||
|
if not path_allowed:
|
||||||
counts["unverifiable_file_count"] += 1
|
counts["unverifiable_file_count"] += 1
|
||||||
continue
|
continue
|
||||||
try:
|
try:
|
||||||
@@ -1320,3 +1334,16 @@ def _is_relative_to(path: Path, root: Path) -> bool:
|
|||||||
except ValueError:
|
except ValueError:
|
||||||
return False
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def _trusted_distribution_root(root: Path) -> bool:
|
||||||
|
for value in sys.path:
|
||||||
|
try:
|
||||||
|
candidate = Path(value or ".").resolve()
|
||||||
|
except OSError:
|
||||||
|
continue
|
||||||
|
if candidate.parent == candidate:
|
||||||
|
continue
|
||||||
|
if root == candidate or _is_relative_to(root, candidate):
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|||||||
Reference in New Issue
Block a user