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
Original file line number Diff line number Diff line change
Expand Up @@ -70,11 +70,18 @@ def _sqlite_path_from_dsn(dsn: str) -> Path:
return Path(dsn)


def _adapter_for_dsn(dsn: str) -> DatabaseAdapter:
"""Select the sync adapter whose backing store matches ``dsn``'s scheme."""
def _adapter_for_dsn(
dsn: str, *, postgres_schema: str | None = None
) -> DatabaseAdapter:
"""Select the sync adapter whose backing store matches ``dsn``'s scheme.

``postgres_schema`` pins the Postgres adapter's search_path for a caller
whose table lives outside ``public`` (OMN-19661). SQLite has no schemas,
so it is ignored there.
"""
scheme = urlsplit(dsn).scheme.lower()
if scheme in _POSTGRES_SCHEMES:
return PostgresSyncProjectionAdapter(dsn)
return PostgresSyncProjectionAdapter(dsn, schema=postgres_schema)
if scheme in _SQLITE_SCHEMES or not scheme:
return SqliteDatabaseAdapter(_sqlite_path_from_dsn(dsn))
raise ValueError(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -299,14 +299,52 @@ def resolve_delegation_claim_store() -> DelegationClaimPort:
return DelegationClaimPort(
resolve_database=lambda: SqliteDatabaseAdapter(default_claim_db_path())
)
return DelegationClaimPort(_adapter_for_dsn(binding.resolve_database_url()))
# OMN-19661: pinned to the schema the contract declares. The table name
# stays bare (the adapter refuses a dotted one), and no role or database on
# a deployed lane sets a search_path, so without the pin every claim here
# resolved to `public`, where the table does not exist.
return DelegationClaimPort(
_adapter_for_dsn(
binding.resolve_database_url(), postgres_schema=claims_schema()
)
)


_CONTRACT_PATH = Path(__file__).resolve().parent.parent / "contract.yaml"


def claims_schema(contract_path: Path = _CONTRACT_PATH) -> str:
"""The schema ``contract.yaml`` declares for :data:`CLAIMS_TABLE`.

Read from ``db_io.db_tables`` rather than restated here, so the relation
the port writes and the relation the migration and the grant name cannot
drift apart. Exactly one entry must declare the table, with a schema;
anything else is refused, because a guessed schema is how a claim lands
in a relation nobody granted or migrated.
"""
import yaml

raw = yaml.safe_load(contract_path.read_text(encoding="utf-8"))
tables = raw.get("db_io", {}).get("db_tables", []) if isinstance(raw, dict) else []
declared = [
entry
for entry in tables
if isinstance(entry, dict) and entry.get("name") == CLAIMS_TABLE
]
if len(declared) != 1 or not str(declared[0].get("schema") or "").strip():
raise ValueError(
f"{contract_path} must declare exactly one db_io.db_tables entry for "
f"{CLAIMS_TABLE!r} with a schema; found {declared!r}"
)
return str(declared[0]["schema"]).strip()


__all__ = [
"CLAIMS_TABLE",
"DelegationClaimPort",
"ModelDelegationClaimOutcome",
"ProtocolDelegationIdempotencyPort",
"claims_schema",
"default_claim_db_path",
"resolve_delegation_claim_store",
]
38 changes: 34 additions & 4 deletions src/omnimarket/projection/postgres_sync_database.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,15 +48,25 @@
# come from trusted internal projection constants and typed row keys (never user
# input), but validating keeps the composed SQL provably injection-free — the
# same posture ``PostgresDataSource`` applies to its table names.
_IDENTIFIER_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
# ``fullmatch``, not ``match``: with ``match`` the ``$`` anchor also accepts a
# trailing newline, so ``"name\n"`` passed the gate (OMN-19661).
_IDENTIFIER_RE = re.compile(r"[a-zA-Z_][a-zA-Z0-9_]*")


def _validate_identifier(name: str, *, kind: str) -> str:
if not _IDENTIFIER_RE.match(name):
if not _IDENTIFIER_RE.fullmatch(name):
raise ValueError(f"invalid {kind} identifier: {name!r}")
return name


def _search_path_option(schema: str) -> str:
"""The libpq ``options`` value pinning ``search_path`` to one schema.

Re-validates, so no caller can build the option from an unchecked name.
"""
return f"-c search_path={_validate_identifier(schema, kind='schema')}"


class PostgresSyncProjectionAdapter:
"""``ProtocolProjectionDatabaseSync`` over a psycopg2 Postgres connection.

Expand All @@ -69,10 +79,21 @@ class PostgresSyncProjectionAdapter:
with a pool).
"""

def __init__(self, dsn: str) -> None:
def __init__(self, dsn: str, *, schema: str | None = None) -> None:
if not dsn.strip():
raise ValueError("PostgresSyncProjectionAdapter requires a non-empty DSN")
self._dsn = dsn
# OMN-19661. Table names reach this adapter BARE (a dotted name is
# refused by the identifier gate here and in the core upsert plan), and
# no role or database on the deployed lanes sets a search_path, so a
# bare name resolves to `public`. A caller whose table lives in another
# schema -- the delegate-skill claim table is in omninode_internal --
# pins it here, and every connection this adapter opens carries it.
# One schema, validated as an identifier, never a list: a search_path
# with a fallback would let a same-named relation elsewhere answer.
self._schema = (
None if schema is None else _validate_identifier(schema, kind="schema")
)

def _connect(self) -> Any:
# This IS the ProtocolProjectionDatabaseSync I/O boundary adapter (the
Expand All @@ -83,7 +104,16 @@ def _connect(self) -> Any:
# annotation, NOT a path-allowlist broadening.
import psycopg2 # type: ignore[import-untyped]

conn = psycopg2.connect(self._dsn) # no-contract-check: projection boundary
if self._schema is None:
conn = psycopg2.connect(self._dsn) # no-contract-check: projection boundary
else:
# The schema passed _validate_identifier in __init__ (letters,
# digits and underscore only, full match), so the libpq options
# string cannot carry a space, quote, comma, newline or a second
# ``-c`` setting.
conn = psycopg2.connect( # no-contract-check: projection boundary
self._dsn, options=_search_path_option(self._schema)
)
conn.autocommit = True
return conn

Expand Down
Loading
Loading