From e26fe4b60882599a6cfaee3bacca81734d27876c Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Fri, 2 Oct 2026 07:26:01 +0200 Subject: [PATCH] fix(memory): the S3 store verifies its bucket and never falls back Maintainer decisions 2026-10-02: "it has the featurs. fallback not desired" and "Admin sets it once; agent verifies". - MinioStore never configures its bucket: configure()/enable_versioning()/ expire() are gone; the client protocol only reads versioning and lifecycle. - verify_bucket() runs when a MinioStore is opened (once per process per store): versioning Enabled, an enabled lifecycle rule expiring /runs/ (none expiring the whole store), S3 Select answering a probe, object tags readable. Every missing piece is listed in one MemoryStoreError with the aws-cli admin command or the permission to grant. - No S3 Select fallback: a server without Select is refused at open, and a failing Select on one object raises (naming RustFS's leading-row schema inference when the field is unbound) instead of filtering locally. - MINIO_REGION is passed through when set. - Tests: fakes for versioning off/unreadable, lifecycle missing/disabled/ tag-narrowed/whole-store, Select missing, tags unreadable, and the pass case; the real-server contract uses OUSAST_MEMORY_TEST_BUCKET (else MINIO_BUCKET) and never writes bucket configuration. - ops/ax/README.md: RustFS support, the admin one-time setup, the agent policy. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- ops/ax/README.md | 52 ++++++++ src/openultrasast/plane/memory.py | 201 ++++++++++++++++++++++-------- tests/test_plane_memory.py | 158 ++++++++++++++++------- 3 files changed, 315 insertions(+), 96 deletions(-) diff --git a/ops/ax/README.md b/ops/ax/README.md index fd65f3f..8b7fb91 100644 --- a/ops/ax/README.md +++ b/ops/ax/README.md @@ -49,6 +49,58 @@ afterwards (`go clean -cache -modcache`); rerunning the full Substrate install o - A Workspace's `files` reach the actor inline in one environment variable: above ~20 KB of content the template fails with "actor template not found" (an 86 KB and a 31 KB excerpt did; 7 KB ran). +## Memory store on S3 (MinIO or RustFS) + +`OUSAST_MEMORY=minio://[/]` puts the plane's memory store (`plane/memory.py`) in an +S3-compatible bucket. MinIO and RustFS both work; the maintainer's server is RustFS. Endpoint and credentials +come from `.env` (`MINIO_ENDPOINT`, `MINIO_ACCESS_KEY`, `MINIO_SECRET_KEY`, `MINIO_SECURE`, optional +`MINIO_REGION`, which avoids a GetBucketLocation call) and are never printed. + +**The store never configures its bucket.** An admin sets it up once. Each time a `MinioStore` is opened it +checks the setup (`MinioStore.verify_bucket`), using reads plus one small probe object at +`/_probe/select.jsonl`. If anything is missing, it refuses to start with a `MemoryStoreError` that lists +each missing piece and the admin command that fixes it. Nothing is skipped quietly. The store needs: + +- **versioning `Enabled`**, because provenance cites object version ids; +- **an enabled lifecycle rule expiring `/runs/`** (raw run outputs) after a number of days, 30 by + default. The rule must be a plain prefix filter, and no rule may expire the whole store (`repos/`, `facts/`, + `index.jsonl` and the blobs are kept); +- **S3 Select** (`SelectObjectContent` over JSON Lines). Every filtered read is pushed down to the server, and + there is no fetch-and-filter fallback. A server without Select (some MinIO releases removed it) is refused; +- **object tags readable**, because rows are filtered by their `kind` tag. + +One-time admin setup (admin credentials, bucket `sast-memory`, store at the bucket root): + + aws s3api put-bucket-versioning --endpoint-url "$MINIO_ENDPOINT" --bucket sast-memory \ + --versioning-configuration Status=Enabled + aws s3api put-bucket-lifecycle-configuration --endpoint-url "$MINIO_ENDPOINT" --bucket sast-memory \ + --lifecycle-configuration '{"Rules":[{"ID":"ousast-runs-expiry","Status":"Enabled", + "Filter":{"Prefix":"runs/"},"Expiration":{"Days":30}}]}' + +`put-bucket-lifecycle-configuration` replaces every rule on the bucket, so merge this rule with any rules +already there. The agent account's policy can read the bucket's configuration but not change it. It reads and +writes objects and their tags, and it gets no `Put*` on versioning or lifecycle: + + {"Version": "2012-10-17", "Statement": [ + {"Effect": "Allow", "Action": ["s3:ListBucket", "s3:GetBucketLocation", "s3:GetBucketVersioning", + "s3:GetLifecycleConfiguration"], + "Resource": ["arn:aws:s3:::sast-memory"]}, + {"Effect": "Allow", "Action": ["s3:GetObject", "s3:GetObjectVersion", "s3:PutObject", "s3:DeleteObject", + "s3:GetObjectTagging", "s3:PutObjectTagging"], + "Resource": ["arn:aws:s3:::sast-memory/*"]}]} + +Measured state on 2026-10-02 (agent `sast-memory-agent`): versioning was `Enabled` and a 30-day rule on `runs/` +was in place. `verify_bucket` still refused, because `GetObjectTagging` returned AccessDenied for the agent. + +Known RustFS limit (measured 2026-10-02): Select infers an object's JSON schema from its leading rows. A +`where` field that is missing there, even if row 5000 has it, fails with `EvaluatorBindingDoesNotExist` instead +of matching. The store raises a `MemoryStoreError` that names the field. It does not answer "no rows", since +that answer could silently drop matches. + +The real-server contract tests (`OUSAST_MEMORY_TEST_MINIO=1`, bucket `OUSAST_MEMORY_TEST_BUCKET`, else +`MINIO_BUCKET`) verify the bucket at its root and write only under a fresh `contract-/` prefix. They never +configure the bucket. + ## Moving to a separate Kubernetes cluster (planned) The maintainer will deploy the plane to its own Kubernetes environment. The manifests under `plane/` move as they diff --git a/src/openultrasast/plane/memory.py b/src/openultrasast/plane/memory.py index 7307d03..ac7dc6b 100644 --- a/src/openultrasast/plane/memory.py +++ b/src/openultrasast/plane/memory.py @@ -12,12 +12,16 @@ engine's own. Layout, identical for both backends:: by excerpt sha, responses/*.json by request sha, programs/*.json ``OUSAST_MEMORY`` selects the backend: ``file:///path`` (:class:`FileStore`, default ``results_root()/memory``) or -``minio://[/]`` (:class:`MinioStore`, the ``minio`` extra). MinIO's endpoint and credentials come -from ``.env`` (``MINIO_ENDPOINT``, ``MINIO_ACCESS_KEY``, ``MINIO_SECRET_KEY``, ``MINIO_SECURE``), never from a -manifest, and are never printed. Ingest is idempotent: an ``index.jsonl`` hit on (run, task, sha256) skips the -delivery, and rows carry a deterministic ``id``, so a repeated row replaces itself. Every object is written whole -(``FileStore``: a temporary file renamed) and the index last, so an interrupted ingest leaves nothing partial that -a repeat would not repair. ``FileStore`` refuses to write below 1 GiB free and names its path. +``minio://[/]`` (:class:`MinioStore`, the ``minio`` extra; MinIO or RustFS). The endpoint and +credentials come from ``.env`` (``MINIO_ENDPOINT``, ``MINIO_ACCESS_KEY``, ``MINIO_SECRET_KEY``, ``MINIO_SECURE``, +``MINIO_REGION``), never from a manifest, and are never printed. The store never configures its bucket: an admin +enables versioning and the ``runs/`` expiry rule once, and opening a ``MinioStore`` verifies them and S3 Select +(:meth:`MinioStore.verify_bucket`), refusing to start otherwise. + +Ingest is idempotent: an ``index.jsonl`` hit on (run, task, sha256) skips the delivery, and rows carry a +deterministic ``id``, so a repeated row replaces itself. Every object is written whole (``FileStore``: a temporary +file renamed) and the index last, so an interrupted ingest leaves nothing partial that a repeat would not repair. +``FileStore`` refuses to write below 1 GiB free and names its path. Fact reuse: a facts entry is valid for (repo, pin, candidates digest, runner image digest), the key the generator writes as the Task annotation ``openultrasast.io/memory-key``. :func:`seed` runs before a Run: for every entry whose @@ -458,8 +462,23 @@ class FileStore(MemoryStore): # --- the MinIO backend --------------------------------------------------------------------------------------------- +@dataclass(frozen=True) +class ExpiryRule: + """One bucket lifecycle rule as :meth:`MinioStore.verify_bucket` reads it: ``prefix`` is the filter's key prefix + (``""`` for a bucket-wide rule), ``tagged`` whether tags also narrow it, ``days`` its current-version + expiration in days (None: no expiration by days), ``expires`` whether it expires current versions at all.""" + + rule_id: str + prefix: str + days: int | None + enabled: bool + tagged: bool = False + expires: bool = True + + class ObjectClient(Protocol): - """The object operations :class:`MinioStore` needs; :class:`SdkClient` adapts the ``minio`` SDK, tests fake it.""" + """The object operations :class:`MinioStore` needs; :class:`SdkClient` adapts the ``minio`` SDK, tests fake it. + The bucket-configuration calls only read: the store never configures its bucket.""" def get(self, key: str) -> tuple[bytes, str | None] | None: ... def put(self, key: str, data: bytes, labels: Mapping[str, str]) -> None: ... @@ -469,21 +488,32 @@ class ObjectClient(Protocol): def version(self, key: str) -> str | None: ... def select(self, key: str, expression: str) -> bytes: ... def presign(self, method: str, key: str, expires: timedelta) -> str: ... - def expire(self, prefix: str, days: int, rule_id: str) -> None: ... - def enable_versioning(self) -> None: ... + def versioning(self) -> str: ... + def lifecycle(self) -> list[ExpiryRule]: ... -_SELECT_PROBES: dict[str, bool] = {} +_VERIFIED: set[str] = set() PROBE_KEY = "_probe/select.jsonl" +RUNS_PREFIX = "runs/" +RUNS_RULE_ID = "ousast-runs-expiry" +RUNS_EXPIRE_DAYS = 30 + + +def _failure(exc: Exception) -> str: + return f"{getattr(exc, 'code', None) or type(exc).__name__}: {str(exc)[:200]}" class MinioStore(MemoryStore): """The layout in one bucket (optionally under a prefix): object metadata and tags on every row object, S3 - Select pushdown when the server answers it (probed once per process) and a local filter otherwise, bucket - versioning for provenance, lifecycle expiry for ``runs/``, presigned URLs.""" + Select pushdown for every filtered read, bucket versioning for provenance, lifecycle expiry for ``runs/``, + presigned URLs. An admin configures the bucket once; opening a store verifies it (:meth:`verify_bucket`) and + refuses a bucket without versioning, the ``runs/`` expiry rule or S3 Select. There is no local fallback.""" def __init__(self, client: ObjectClient, bucket: str, prefix: str = "") -> None: self.client, self.bucket, self.prefix = client, bucket, prefix.strip("/") + if self.describe() not in _VERIFIED: + self.verify_bucket() + _VERIFIED.add(self.describe()) def describe(self) -> str: return f"minio://{self.bucket}" + (f"/{self.prefix}" if self.prefix else "") @@ -511,32 +541,96 @@ class MinioStore(MemoryStore): tagged = self.client.tags(self._k(key)).get("kind") return tagged is None or tagged == "*" or kind in tagged.split() - def select_supported(self) -> bool: - """Whether the server answers S3 Select over JSON Lines: one probe per bucket per process.""" - name = self.describe() - if name not in _SELECT_PROBES: - try: - self.client.put(self._k(PROBE_KEY), b'{"probe":1}\n{"probe":2}\n', {}) - got = _parse_lines(self.client.select(self._k(PROBE_KEY), where_sql({"probe": 1})), PROBE_KEY) - _SELECT_PROBES[name] = got == [{"probe": 1}] - detail = "" if _SELECT_PROBES[name] else f"; the probe returned {got!r}" - except Exception as exc: # noqa: BLE001 -- any failure means "not supported"; the fallback is exact - _SELECT_PROBES[name], detail = False, f"; {type(exc).__name__}: {str(exc)[:200]}" - if not _SELECT_PROBES[name]: - log.warning("memory %s: the server does not answer S3 Select%s; filtering rows locally", name, detail) - return _SELECT_PROBES[name] + def verify_bucket(self) -> None: + """Read-only check that the admin's one-time setup is in place: versioning ``Enabled``, an enabled + lifecycle rule expiring ``/runs/`` and none expiring the whole store, S3 Select answering a probe + over JSON Lines, and the probe's object tags readable. Every missing piece is named with its admin action + in one :class:`MemoryStoreError`.""" + runs = self._k(RUNS_PREFIX) + endpoint = '--endpoint-url "$MINIO_ENDPOINT"' + missing: list[str] = [] + try: + status = self.client.versioning() + except Exception as exc: # noqa: BLE001 -- any failure is named; the store does not start + missing.append( + f"versioning cannot be read ({_failure(exc)}): the agent needs s3:GetBucketVersioning on " + f"arn:aws:s3:::{self.bucket}, and versioning must be Enabled" + ) + else: + if status != "Enabled": + missing.append( + f"versioning is {status or 'Off'}, it must be Enabled (provenance cites object versions). Admin: " + f"aws s3api put-bucket-versioning {endpoint} --bucket {self.bucket} " + "--versioning-configuration Status=Enabled" + ) + rule_json = ( + f'{{"Rules":[{{"ID":"{RUNS_RULE_ID}","Status":"Enabled","Filter":{{"Prefix":"{runs}"}},' + f'"Expiration":{{"Days":{RUNS_EXPIRE_DAYS}}}}}]}}' + ) + lifecycle_admin = ( + f"Admin (merged with any rules the bucket already has): aws s3api put-bucket-lifecycle-configuration " + f"{endpoint} --bucket {self.bucket} --lifecycle-configuration '{rule_json}'" + ) + try: + rules = self.client.lifecycle() + except Exception as exc: # noqa: BLE001 + missing.append( + f"the lifecycle configuration cannot be read ({_failure(exc)}): the agent needs " + f"s3:GetLifecycleConfiguration on arn:aws:s3:::{self.bucket}, and a rule must expire {runs}" + ) + else: + live = [r for r in rules if r.enabled and r.expires] + store_root = self._k("") + wide = [r.rule_id for r in live if store_root.startswith(r.prefix)] + if wide: + missing.append( + f"lifecycle rule(s) {', '.join(wide)} expire the whole store (repos/, facts/ and the index would be " + f"deleted); limit them to {runs}" + ) + if not any(r.prefix in (runs, runs.rstrip("/")) and not r.tagged and r.days for r in live): + missing.append(f"no enabled lifecycle rule expires {runs} after a number of days. {lifecycle_admin}") + try: + self.client.put(self._k(PROBE_KEY), b'{"probe":1}\n{"probe":2}\n', {}) + got = _parse_lines(self.client.select(self._k(PROBE_KEY), where_sql({"probe": 1})), PROBE_KEY) + select_error = "" if got == [{"probe": 1}] else f"the probe returned {got!r}" + except Exception as exc: # noqa: BLE001 + select_error = _failure(exc) + try: + self.client.tags(self._k(PROBE_KEY)) + except Exception as exc: # noqa: BLE001 + missing.append( + f"object tags cannot be read on {self._k(PROBE_KEY)} ({_failure(exc)}): the store filters row objects by " + f"their kind tag, so the agent needs s3:GetObjectTagging on arn:aws:s3:::{self.bucket}/*" + ) + if select_error: + missing.append( + f"S3 Select (SelectObjectContent over JSON Lines) does not answer on {self._k(PROBE_KEY)} " + f"({select_error}). The store requires it and has no local fallback: use a server with S3 Select " + "(RustFS has it; some MinIO releases removed it) and let the agent s3:PutObject and s3:GetObject" + ) + if missing: + raise MemoryStoreError( + f"memory {self.describe()}: the bucket is not set up for the store. The store never configures its " + "bucket (an admin sets it once, the store verifies); missing:\n" + "\n".join(f" - {m}" for m in missing) + ) def _select(self, key: str, where: Mapping[str, Any]) -> tuple[list[dict[str, Any]], str | None] | None: - if not where or not self.select_supported(): - return super()._select(key, where) + if not where: + return super()._select(key, where) # no filter: the whole object is the answer version = self.client.version(self._k(key)) # stat, then select: Select takes no version id if version is None and self.client.get(self._k(key)) is None: return None try: return _parse_lines(self.client.select(self._k(key), where_sql(where)), key), version - except Exception as exc: # noqa: BLE001 -- one object's Select failing never loses rows - log.warning("memory %s: S3 Select on %s failed (%s); filtering it locally", self.describe(), key, type(exc).__name__) - return super()._select(key, where) + except Exception as exc: + hint = "" + if getattr(exc, "code", None) == "EvaluatorBindingDoesNotExist": + # RustFS infers an object's JSON schema from its leading rows: a field missing there is unbound even + # when later rows carry it, so "no rows" would be a guess. Measured 2026-10-02 (field at row 5000). + hint = f" (a field of {sorted(where)} is absent from the object's leading rows, which the server reads as its schema)" + raise MemoryStoreError( + f"memory {self.describe()}: S3 Select on {key} failed ({_failure(exc)}){hint}; the store has no local fallback" + ) from exc def presign_put(self, key: str, expires: timedelta = timedelta(hours=1)) -> str: return self.client.presign("PUT", self._k(key), expires) @@ -544,24 +638,17 @@ class MinioStore(MemoryStore): def presign_get(self, key: str, expires: timedelta = timedelta(hours=1)) -> str: return self.client.presign("GET", self._k(key), expires) - def configure(self, runs_expire_days: int = 30) -> None: - """Bucket versioning on (provenance) and the lifecycle rule expiring ``runs/`` after ``runs_expire_days``; - ``repos/``, ``facts/`` and ``proposals/`` are kept.""" - if runs_expire_days < 1: - raise ValueError("runs_expire_days must be at least 1") - self.client.enable_versioning() - self.client.expire(self._k("runs/"), runs_expire_days, "ousast-runs-expiry") - class SdkClient: """:class:`ObjectClient` over ``minio.Minio``; imported lazily so the core install needs no SDK.""" - def __init__(self, endpoint: str, access_key: str, secret_key: str, secure: bool, bucket: str) -> None: + def __init__(self, endpoint: str, access_key: str, secret_key: str, secure: bool, bucket: str, region: str | None = None) -> None: try: from minio import Minio except ImportError as exc: raise MemoryStoreError("OUSAST_MEMORY=minio://... needs the minio SDK: install openultrasast[minio]") from exc - self.sdk, self.bucket = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=secure), bucket + self.sdk = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=secure, region=region) + self.bucket = bucket def get(self, key: str) -> tuple[bytes, str | None] | None: from minio.error import S3Error @@ -624,18 +711,22 @@ class SdkClient: return str(self.sdk.presigned_put_object(self.bucket, key, expires=expires)) return str(self.sdk.presigned_get_object(self.bucket, key, expires=expires)) - def expire(self, prefix: str, days: int, rule_id: str) -> None: - from minio.commonconfig import ENABLED, Filter - from minio.lifecycleconfig import Expiration, LifecycleConfig, Rule + def versioning(self) -> str: + return str(self.sdk.get_bucket_versioning(self.bucket).status_string) - rule = Rule(ENABLED, rule_filter=Filter(prefix=prefix), rule_id=rule_id, expiration=Expiration(days=days)) - self.sdk.set_bucket_lifecycle(self.bucket, LifecycleConfig([rule])) - - def enable_versioning(self) -> None: - from minio.commonconfig import ENABLED - from minio.versioningconfig import VersioningConfig - - self.sdk.set_bucket_versioning(self.bucket, VersioningConfig(ENABLED)) + def lifecycle(self) -> list[ExpiryRule]: + config = self.sdk.get_bucket_lifecycle(self.bucket) + rules: list[ExpiryRule] = [] + for rule in config.rules if config else []: + found = rule.rule_filter + joined = found.and_operator if found else None + prefix = (joined.prefix if joined else found.prefix if found else None) or "" + tagged = bool(found and (found.tag or (joined and joined.tags))) + expiration = rule.expiration + expires = bool(expiration and (expiration.days or expiration.date)) + days = expiration.days if expiration else None + rules.append(ExpiryRule(str(rule.rule_id or ""), prefix, days, rule.status == "Enabled", tagged, expires)) + return rules def minio_settings(environ: Mapping[str, str] | None = None) -> dict[str, Any]: @@ -649,10 +740,13 @@ def minio_settings(environ: Mapping[str, str] | None = None) -> dict[str, Any]: if missing: raise MemoryStoreError(f"OUSAST_MEMORY=minio://... needs {', '.join(missing)} in .env or the environment") secure = (environ.get("MINIO_SECURE") or "true").strip().lower() not in ("0", "false", "no", "off") - return { + settings: dict[str, Any] = { "endpoint": environ["MINIO_ENDPOINT"], "access_key": environ["MINIO_ACCESS_KEY"], "secret_key": environ["MINIO_SECRET_KEY"], "secure": secure, } # fmt: skip + if environ.get("MINIO_REGION"): + settings["region"] = environ["MINIO_REGION"] # spares a GetBucketLocation the agent may not be allowed + return settings def open_store(spec: str | None = None, environ: Mapping[str, str] | None = None) -> MemoryStore: @@ -766,6 +860,7 @@ __all__ = [ "MemoryStoreError", "MinioStore", "ObjectClient", + "ExpiryRule", "Record", "SdkClient", "counts", diff --git a/tests/test_plane_memory.py b/tests/test_plane_memory.py index e16d3d6..655f492 100644 --- a/tests/test_plane_memory.py +++ b/tests/test_plane_memory.py @@ -1,5 +1,6 @@ """The plane's memory store (harnessx-removal Req 6.1/6.4, design ยง4): one contract over every backend, the S3 -Select probe and its fallback on a fake object client, backend selection, fact reuse by ``seed``, disk refusal.""" +bucket verification (versioning, the ``runs/`` expiry rule, S3 Select; no fallback) on a fake object client, backend +selection, fact reuse by ``seed``, disk refusal.""" from __future__ import annotations @@ -20,6 +21,7 @@ import pytest from openultrasast.plane import memory, reconciler from openultrasast.plane.memory import ( MINIO_KEYS, + ExpiryRule, FileStore, MemoryStore, MemoryStoreError, @@ -39,8 +41,9 @@ OTHER_PIN = "fedcba9876543210fedcba9876543210fedcba98" IMAGE = "sha256:" + "a" * 64 FACTS = json.dumps({"callers": {"a.py::run": []}, "counts": {"files": 1}, "files": {"a.py": {"functions": ["run"]}}}).encode() MINIO_SKIP = ( - "MinIO contract tests need OUSAST_MEMORY_TEST_MINIO=1 plus MINIO_ENDPOINT, MINIO_ACCESS_KEY and MINIO_SECRET_KEY " - "exported (optional: MINIO_SECURE, OUSAST_MEMORY_TEST_BUCKET, default ousast-memory-test) and the minio extra installed" + "MinIO/RustFS contract tests need OUSAST_MEMORY_TEST_MINIO=1 plus MINIO_ENDPOINT, MINIO_ACCESS_KEY, MINIO_SECRET_KEY " + "and a bucket (OUSAST_MEMORY_TEST_BUCKET, else MINIO_BUCKET) exported (optional: MINIO_SECURE, MINIO_REGION) and the " + "minio extra installed; the admin configures that bucket, the tests never do" ) @@ -54,12 +57,18 @@ def verdict(candidate: str, final: str, **fields: Any) -> dict[str, Any]: return row("verdict", candidate, candidate=candidate, family="injection", final=final, tiebreak=False, **fields) +# The admin's rule for the prefixes these tests open stores under ("" and "p"). +RUNS_RULES = [ExpiryRule("ousast-runs-expiry", "runs/", 30, True), ExpiryRule("ousast-runs-expiry-p", "p/runs/", 30, True)] + + class FakeObjects: """An in-memory :class:`memory.ObjectClient`: versions per write, tags per object; ``select`` works, raises or - answers wrongly, evaluating the equality clauses :func:`where_sql` writes.""" + answers wrongly, evaluating the equality clauses :func:`where_sql` writes. The bucket is configured as the admin + does it unless ``versioning``/``rules`` say otherwise (``"denied"``: the read is refused).""" - def __init__(self, select: str = "works") -> None: + def __init__(self, select: str = "works", versioning: str = "Enabled", rules: list[ExpiryRule] | str | None = None) -> None: self.mode, self.objects, self.versions = select, {}, {} + self.status, self.rules, self.tags_denied = versioning, RUNS_RULES if rules is None else rules, False self.selects: list[str] = [] self.gets: list[str] = [] self.calls: list[tuple[Any, ...]] = [] @@ -80,6 +89,8 @@ class FakeObjects: return sorted(k for k in self.objects if k.startswith(prefix)) def tags(self, key: str) -> dict[str, str]: + if self.tags_denied: + raise PermissionError("Access Denied") return dict(self.objects[key][1]) def version(self, key: str) -> str | None: @@ -91,6 +102,8 @@ class FakeObjects: raise RuntimeError("NotImplemented: SelectObjectContent is not supported") if self.mode == "wrong": return b"" + if self.mode == "unbound": # RustFS: a field absent from the object's leading rows + raise type("S3Error", (Exception,), {"code": "EvaluatorBindingDoesNotExist"})("A column name ... does not exist") clauses = re.findall(r"s\.\"(\w+)\" = ('(?:[^']|'')*'|true|false|-?[\d.]+)", expression) want = {n: v[1:-1].replace("''", "'") if v.startswith("'") else json.loads(v) for n, v in clauses} rows = [json.loads(line) for line in self.objects[key][0].splitlines() if line.strip()] @@ -99,41 +112,49 @@ class FakeObjects: def presign(self, method: str, key: str, expires: timedelta) -> str: return f"https://minio.test/{key}?method={method}&X-Amz-Expires={int(expires.total_seconds())}" - def expire(self, prefix: str, days: int, rule_id: str) -> None: - self.calls.append(("expire", prefix, days, rule_id)) + def versioning(self) -> str: + if self.status == "denied": + raise PermissionError("Access Denied") + return self.status - def enable_versioning(self) -> None: - self.calls.append(("versioning",)) + def lifecycle(self) -> list[ExpiryRule]: + if self.rules == "denied": + raise PermissionError("Access Denied") + return list(self.rules) @pytest.fixture(autouse=True) -def fresh_probes(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(memory, "_SELECT_PROBES", {}) +def fresh_verification(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(memory, "_VERIFIED", set()) def _real_minio() -> Iterator[MemoryStore]: if os.environ.get("OUSAST_MEMORY_TEST_MINIO") != "1" or any(not os.environ.get(k) for k in MINIO_KEYS): pytest.skip(MINIO_SKIP) pytest.importorskip("minio", reason=MINIO_SKIP) - bucket = os.environ.get("OUSAST_MEMORY_TEST_BUCKET") or "ousast-memory-test" + bucket = os.environ.get("OUSAST_MEMORY_TEST_BUCKET") or os.environ.get("MINIO_BUCKET") + if not bucket: + pytest.skip(MINIO_SKIP) client = SdkClient(bucket=bucket, **minio_settings(os.environ)) - if not client.sdk.bucket_exists(bucket): - client.sdk.make_bucket(bucket) - store = MinioStore(client, bucket, f"contract-{uuid.uuid4().hex[:12]}") - client.enable_versioning() + MinioStore(client, bucket) # verifies the admin's setup at the bucket root; the tests never configure the bucket + # The contract rows go under a fresh prefix so every test starts empty; the admin's expiry rule covers the root's + # runs/, not this prefix's, so the prefixed store counts as verified by the root's check above. + prefix = f"contract-{uuid.uuid4().hex[:12]}" + memory._VERIFIED.add(f"minio://{bucket}/{prefix}") + store = MinioStore(client, bucket, prefix) yield store for key in client.keys(store.prefix + "/"): client.delete(key) -@pytest.fixture(params=["file", "fake-minio-select", "fake-minio-fallback", "minio"]) +@pytest.fixture(params=["file", "fake-minio", "minio"]) def store(request: pytest.FixtureRequest, tmp_path: Path) -> Iterator[MemoryStore]: if request.param == "file": yield FileStore(tmp_path / "memory") elif request.param == "minio": yield from _real_minio() else: - yield MinioStore(FakeObjects("works" if request.param.endswith("select") else "raises"), "bucket", "p") + yield MinioStore(FakeObjects(), "bucket", "p") # --- the contract -------------------------------------------------------------------------------------------------- @@ -215,33 +236,92 @@ def test_select_is_pushed_down_when_the_probe_answers(caplog: pytest.LogCaptureF store = _filled(client) with caplog.at_level(logging.WARNING, logger="openultrasast.plane.memory"): got = [r.row["candidate"] for r in store.rows(kind="verdict", where={"final": "disputed"})] - assert got == ["a.py::x"] and store.select_supported() and "locally" not in caplog.text + assert got == ["a.py::x"] and "locally" not in caplog.text assert client.selects[0] == 'SELECT * FROM S3Object s WHERE s."probe" = 1' assert client.selects[1:] == ["SELECT * FROM S3Object s WHERE s.\"final\" = 'disputed' AND s.\"kind\" = 'verdict'"] @pytest.mark.parametrize("mode", ["raises", "wrong"]) -def test_without_select_the_same_rows_come_from_a_local_filter_logged_once( - mode: str, caplog: pytest.LogCaptureFixture, tmp_path: Path -) -> None: - client = FakeObjects(mode) - store, reference = _filled(client), FileStore(tmp_path / "m") - reference.put_rows([r.row for r in store.rows()]) - with caplog.at_level(logging.WARNING, logger="openultrasast.plane.memory"): - for where in ({"final": "disputed"}, {"final": "agreed"}, {"pass": "a"}): - assert [r.row for r in store.rows(where=where)] == [r.row for r in reference.rows(where=where)] != [] - assert not store.select_supported() and len(client.selects) == 1, "one probe per process" - assert caplog.text.count("does not answer S3 Select") == 1 +def test_a_server_without_select_is_refused_at_open(mode: str) -> None: + with pytest.raises(MemoryStoreError, match=r"S3 Select .* does not answer on _probe/select.jsonl .*no local fallback") as caught: + MinioStore(FakeObjects(mode), "bucket") + assert "versioning" not in str(caught.value) and "lifecycle" not in str(caught.value) -def test_a_failing_select_on_one_object_falls_back_for_that_object(caplog: pytest.LogCaptureFixture) -> None: +def test_a_failing_select_on_one_object_raises_instead_of_filtering_locally() -> None: client = FakeObjects("works") store = _filled(client) - assert store.select_supported() client.mode = "raises" - with caplog.at_level(logging.WARNING, logger="openultrasast.plane.memory"): - assert [r.row["candidate"] for r in store.rows(where={"final": "agreed"})] == ["a.py::run"] - assert "filtering it locally" in caplog.text + with pytest.raises(MemoryStoreError, match=r"S3 Select on repos/github.com__o__r/\w+\.jsonl failed .*no local fallback"): + store.rows(where={"final": "agreed"}) + + +def test_an_unbound_field_on_rustfs_raises_naming_the_schema_inference_never_returning_no_rows() -> None: + client = FakeObjects("works") + store = _filled(client) + client.mode = "unbound" + with pytest.raises(MemoryStoreError, match=r"EvaluatorBindingDoesNotExist.*\['usd'\] is absent from the object's leading rows"): + store.rows(where={"usd": 0.5}) + + +# --- bucket verification: the admin configures once, the store only reads ------------------------------------------ + + +def test_verify_bucket_passes_on_the_admin_setup_and_writes_only_its_probe() -> None: + client = FakeObjects() + store = MinioStore(client, "bucket", "p") + assert sorted(client.objects) == ["p/_probe/select.jsonl"] and not hasattr(client, "enable_versioning") + store.verify_bucket() + MinioStore(client, "bucket", "p") + assert len(client.selects) == 2, "opening a verified store again does not re-verify in the same process" + + +@pytest.mark.parametrize("status", ["Suspended", "Off"]) +def test_verify_bucket_refuses_versioning_off_naming_the_admin_command(status: str) -> None: + with pytest.raises(MemoryStoreError, match=rf"versioning is {status}, it must be Enabled") as caught: + MinioStore(FakeObjects(versioning=status), "sast-memory") + message = str(caught.value) + assert "never configures its bucket" in message and "lifecycle" not in message and "Select" not in message + assert 'aws s3api put-bucket-versioning --endpoint-url "$MINIO_ENDPOINT" --bucket sast-memory' in message + + +@pytest.mark.parametrize( + "rules", + [ + [], + [ExpiryRule("other", "logs/", 7, True)], + [ExpiryRule("ousast-runs-expiry", "runs/", 30, False)], + [ExpiryRule("ousast-runs-expiry", "runs/", 30, True, tagged=True)], + [ExpiryRule("ousast-runs-expiry", "runs/", None, True, expires=False)], + ], + ids=["none", "other-prefix", "disabled", "tag-narrowed", "no-expiration"], +) +def test_verify_bucket_refuses_a_missing_runs_expiry_rule(rules: list[ExpiryRule]) -> None: + with pytest.raises(MemoryStoreError, match=r"no enabled lifecycle rule expires runs/") as caught: + MinioStore(FakeObjects(rules=rules), "sast-memory") + assert "put-bucket-lifecycle-configuration" in str(caught.value) + assert '"Filter":{"Prefix":"runs/"},"Expiration":{"Days":30}' in str(caught.value) + + +def test_verify_bucket_names_the_prefixed_runs_and_refuses_a_rule_expiring_the_whole_store() -> None: + with pytest.raises(MemoryStoreError, match=r"no enabled lifecycle rule expires mem/runs/"): + MinioStore(FakeObjects(rules=[ExpiryRule("r", "runs/", 30, True)]), "bucket", "mem") + wide = [ExpiryRule("everything", "", 30, True), ExpiryRule("ousast-runs-expiry", "runs/", 30, True)] + with pytest.raises(MemoryStoreError, match=r"lifecycle rule\(s\) everything expire the whole store"): + MinioStore(FakeObjects(rules=wide), "bucket") + MinioStore(FakeObjects(rules=[ExpiryRule("r", "mem/runs/", 14, True)]), "bucket", "mem") + + +def test_verify_bucket_names_every_missing_piece_and_the_permissions_to_read_them() -> None: + client = FakeObjects("raises", versioning="denied", rules="denied") + client.tags_denied = True + with pytest.raises(MemoryStoreError) as caught: + MinioStore(client, "sast-memory") + message = str(caught.value) + assert "versioning cannot be read (PermissionError: Access Denied)" in message and "s3:GetBucketVersioning" in message + assert "lifecycle configuration cannot be read" in message and "s3:GetLifecycleConfiguration" in message + assert "object tags cannot be read" in message and "s3:GetObjectTagging" in message + assert "S3 Select" in message and message.count("\n - ") == 4 def test_row_objects_carry_tags_and_a_kind_tag_skips_objects_without_that_kind() -> None: @@ -256,14 +336,6 @@ def test_row_objects_carry_tags_and_a_kind_tag_skips_objects_without_that_kind() assert f"repos/github.com__o__r/{PIN}.jsonl" not in client.gets, "an object tagged without the kind is not read" -def test_configure_turns_versioning_on_and_expires_runs_only() -> None: - client = FakeObjects() - MinioStore(client, "bucket", "memory").configure(14) - assert client.calls == [("versioning",), ("expire", "memory/runs/", 14, "ousast-runs-expiry")] - with pytest.raises(ValueError): - MinioStore(client, "bucket").configure(0) - - def test_where_sql_quotes_values_and_refuses_odd_fields() -> None: assert where_sql({}) == "SELECT * FROM S3Object s" assert where_sql({"a": "it's", "b": True, "c": None, "d": 2}) == (