marius/Open_Ultra_SAST
1
0
Fork 0
mirror of https://github.com/norandom/OpenUltraSAST.git synced 2026-10-16 03:49:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

merge(memory): the S3 store verifies its bucket and never falls back

Browse Source 
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
2026-10-02 07:33:44 +02:00
co-authored by Claude Opus 5.5
parent 78c31ba722 e26fe4b608
commit 30cde99e2d
3 changed files with 315 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ops/ax/README.md
+52
View File
@@ -117,6 +117,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://<bucket>[/<prefix>]` 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
`<prefix>/_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 `<prefix>/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-<id>/` 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
src/openultrasast/plane/memory.py
+145 -50
View File
@@ -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://<bucket>[/<prefix>]`` (: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://<bucket>[/<prefix>]`` (: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:
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 ``<prefix>/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_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]
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",
tests/test_plane_memory.py
+115 -43
View File
@@ -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}) == (