Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 76 additions & 0 deletions apps/api/src/cora/infrastructure/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from cora.infrastructure.auth.config import IdentityProviderConfig
from cora.infrastructure.control_port_route import ControlPortRoute
from cora.shared.capture_phase import CapturePhase
from cora.shared.storage_root import normalize_storage_root

_ALLOWED_DATABASE_SCHEMES = ("postgresql://", "postgres://")

Expand Down Expand Up @@ -1217,6 +1218,81 @@ def _validate_scan_probe_remote_python(
raise ValueError(msg)
return value

@field_validator("scan_probe_remote_host")
@classmethod
def _validate_scan_probe_remote_host(cls, value: str | None) -> str | None:
"""Reject an empty or whitespace-only host rather than silently
treating it as unset.

`SCAN_PROBE_REMOTE_HOST=""` is a different
signal than the variable being absent -- in a deployment's
settings template it usually means an interpolation that
resolved to nothing -- and `active_scan_transport` tests
`is not None`, so a bare "" was passing through as a configured
remote transport with no host to connect to, then failing the
vault's own CHECK constraint on the first upsert instead of at
boot. Coercing "" to `None` was the other option; rejecting is
chosen instead so the misconfiguration surfaces immediately
rather than being silently papered over. Leave the setting
unset to mean "no remote probe."
"""
if value is not None and not value.strip():
msg = (
"scan_probe_remote_host is set to an empty or "
"whitespace-only string. Leave it unset to disable the "
"remote scan probe, rather than setting it to an empty value."
)
raise ValueError(msg)
return value

@field_validator("posix_checksum_roots")
@classmethod
def _validate_posix_checksum_roots(cls, value: tuple[str, ...]) -> tuple[str, ...]:
"""Refuse a root that cannot name a real storage tier at boot,
not at the first attestation. A relative path is meaningless to
`PosixChecksumAdapter`, which treats every root as absolute, and
`normalize_storage_root("/")` collapses to the empty string,
which the run_capture_path vault's own CHECK constraint forbids
-- so bare "/" would otherwise pass here and fail on the first
write instead of at boot. A trailing slash is fine: normalization
(`cora.shared.storage_root`) handles it.
"""
for root in value:
if not root.startswith("/"):
msg = f"posix_checksum_roots entry {root!r} is not an absolute path."
raise ValueError(msg)
if not normalize_storage_root(root):
msg = (
f"posix_checksum_roots entry {root!r} normalizes to the "
"empty string. A root must name a facility-level storage "
"tier, not the filesystem root itself."
)
raise ValueError(msg)
return value

@field_validator("scan_probe_allowed_roots")
@classmethod
def _validate_scan_probe_allowed_roots(cls, value: tuple[str, ...]) -> tuple[str, ...]:
"""Same rule as `posix_checksum_roots`, applied to the roots
allowlisted on `scan_probe_remote_host`: a relative path or a
bare "/" is a misconfiguration worth refusing at boot rather
than discovering when the remote probe's own resolver refuses
every locator. A trailing slash is fine: normalization
(`cora.shared.storage_root`) handles it.
"""
for root in value:
if not root.startswith("/"):
msg = f"scan_probe_allowed_roots entry {root!r} is not an absolute path."
raise ValueError(msg)
if not normalize_storage_root(root):
msg = (
f"scan_probe_allowed_roots entry {root!r} normalizes to the "
"empty string. A root must name a facility-level storage "
"tier, not the filesystem root itself."
)
raise ValueError(msg)
return value

@field_validator("capture_experiment_identity_pvs")
@classmethod
def _validate_capture_experiment_identity_pvs(
Expand Down
98 changes: 98 additions & 0 deletions apps/api/tests/unit/test_settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -623,3 +623,101 @@ def test_settings_scan_probe_remote_host_with_remote_python_is_accepted(
settings = Settings()
assert settings.scan_probe_remote_host == "tomdet"
assert settings.scan_probe_remote_python == "/venv/bin/python3"


@pytest.mark.unit
def test_settings_scan_probe_remote_host_rejects_empty_string(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""`SCAN_PROBE_REMOTE_HOST=""` used to pass validation and then
`active_scan_transport` (`is not None`) treated it as a configured
remote host, failing the vault's CHECK constraint on first upsert
instead of at boot."""
import pydantic

monkeypatch.setenv("SCAN_PROBE_REMOTE_HOST", "")
with pytest.raises(pydantic.ValidationError, match="scan_probe_remote_host is set to"):
Settings()


@pytest.mark.unit
def test_settings_scan_probe_remote_host_rejects_whitespace_only(
monkeypatch: pytest.MonkeyPatch,
) -> None:
import pydantic

monkeypatch.setenv("SCAN_PROBE_REMOTE_HOST", " ")
with pytest.raises(pydantic.ValidationError, match="scan_probe_remote_host is set to"):
Settings()


# ---------------------------------------------------------------------------
# posix_checksum_roots / scan_probe_allowed_roots: absolute-path boot checks
# ---------------------------------------------------------------------------


@pytest.mark.unit
def test_settings_posix_checksum_roots_accepts_trailing_slash(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Normalization (`cora.shared.storage_root`) handles the trailing
slash; the validator must not reject what normalization already fixes."""
monkeypatch.setenv("POSIX_CHECKSUM_ROOTS", '["/local1/2BM/"]')
settings = Settings()
assert settings.posix_checksum_roots == ("/local1/2BM/",)


@pytest.mark.unit
def test_settings_posix_checksum_roots_rejects_relative_path(
monkeypatch: pytest.MonkeyPatch,
) -> None:
import pydantic

monkeypatch.setenv("POSIX_CHECKSUM_ROOTS", '["local1/2BM"]')
with pytest.raises(pydantic.ValidationError, match="is not an absolute path"):
Settings()


@pytest.mark.unit
def test_settings_posix_checksum_roots_rejects_bare_root(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A bare "/" normalizes to the empty string, which the
run_capture_path vault's CHECK constraint forbids; refuse it at
boot instead of at the first write."""
import pydantic

monkeypatch.setenv("POSIX_CHECKSUM_ROOTS", '["/"]')
with pytest.raises(pydantic.ValidationError, match="normalizes to the empty string"):
Settings()


@pytest.mark.unit
def test_settings_scan_probe_allowed_roots_accepts_trailing_slash(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("SCAN_PROBE_ALLOWED_ROOTS", '["/local1/2BM/"]')
settings = Settings()
assert settings.scan_probe_allowed_roots == ("/local1/2BM/",)


@pytest.mark.unit
def test_settings_scan_probe_allowed_roots_rejects_relative_path(
monkeypatch: pytest.MonkeyPatch,
) -> None:
import pydantic

monkeypatch.setenv("SCAN_PROBE_ALLOWED_ROOTS", '["local1/2BM"]')
with pytest.raises(pydantic.ValidationError, match="is not an absolute path"):
Settings()


@pytest.mark.unit
def test_settings_scan_probe_allowed_roots_rejects_bare_root(
monkeypatch: pytest.MonkeyPatch,
) -> None:
import pydantic

monkeypatch.setenv("SCAN_PROBE_ALLOWED_ROOTS", '["/"]')
with pytest.raises(pydantic.ValidationError, match="normalizes to the empty string"):
Settings()
Loading