bitflicker64 commented on code in PR #380: URL: https://github.com/apache/hugegraph-ai/pull/380#discussion_r4332840297
########## server/ontogeny/policy/engine.py: ########## @@ -0,0 +1,167 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Cedar-subset policy engine (default driver). + +Parses the ``permit(...) when { ... }`` statements used by PolicySet files: + + permit(principal, action == Action::"close-work-order", resource); + permit(principal in Role::"maintenance_supervisor", + action == Action::"close-work-order", resource) + when { resource.status != "CLOSED" && principal has site && resource.site == principal.site }; + +Evaluation is deterministic and default-deny. Conditions are translated to the +platform mini-expr (``&&``->``and``) so there is exactly ONE expression +semantics in the system. If the ``cedarpy`` binding is installed it is +preferred; this pure-Python driver is the portable fallback and CI reference. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass + +from ..errors import ExpressionError, PolicyDeniedError +from ..core import expr as _expr + +_PERMIT_RE = re.compile( + r"permit\s*\(\s*(?P<principal>[^,]+?)\s*,\s*action\s*==\s*Action::\"(?P<action>[^\"]+)\"\s*,\s*(?P<resource>[^)]*?)s*\)\s*(?:when\s*\{(?P<when>.*?)\}\s*)?;", + re.DOTALL, +) + + +@dataclass(frozen=True) +class PermitRule: + role: str | None # None = any authenticated principal + action: str + when: str | None # mini-expr after translation + + +@dataclass(frozen=True) +class PolicyDecision: + allow: bool + rule: PermitRule | None = None + reason: str = "" + + +def translate_when(cond: str) -> str: + """Cedar condition -> mini-expr: ``&&``/``||`` to and/or, ``resource`` to + ``target`` (the platform's single expression semantics).""" + out = re.sub(r"\bresource\b", "target", cond) + return out.replace("&&", " and ").replace("||", " or ") + + +def parse_policy_set(text: str) -> list[PermitRule]: + rules: list[PermitRule] = [] + for m in _PERMIT_RE.finditer(text): Review Comment: Important: `forbid` statements are dropped without any error, so a policy author's deny never takes effect. `parse_policy_set` only iterates `_PERMIT_RE` matches, and `decide()` returns on the first matching permit. Take `permit(principal in Role::"planner", action == Action::"close", resource);` followed by `forbid(principal, action == Action::"close", resource) when { resource.status == "CLOSED" };`. At this head `parse_policy_set` returns only the permit, so planners can still close CLOSED rows. In Cedar a matching forbid always overrides a permit, and the validator does not flag the ignored statement. The docstrings here and in `policy/__init__.py` say an installed `cedarpy` takes precedence, and `pyproject.toml` ships a `cedar = ["cedarpy>=0.5"]` extra, but nothing under `server/` imports `cedarpy`, so installing it changes nothing. Please either evaluate `forbid` (deny when any matching forbid holds, checked before permits), or make the validator reject any statement in a `. cedar` file that this parser does not consume (`forbid`, `unless`, unsupported clauses). Either way an unsupported rule then fails at publish instead of being dropped. ########## server/ontogeny/api/schemas.py: ########## @@ -0,0 +1,274 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Request bodies for the REST surface. Module-level on purpose: FastAPI +resolves handler annotations against the module namespace -- a model class +nested inside a function silently degrades the parameter to a query param.""" +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel + +# ------------------------------------------------------------------ schemas + + +class QueryBody(BaseModel): + filter: Any = None + sort: Any = None + limit: int = 100 Review Comment: Important: Unauthenticated reads have no upper bound. `QueryBody.limit` is an unbounded `int` that reaches `repo.query(...).limit(limit)` unchanged. An anonymous `POST /api/v1/objects/{type}/query` with `{"limit": 100000000, "links": [...]}` therefore loads the whole table, then runs one link lookup plus one `get_many` per returned row and link (`QueryService.query`). `TraverseBody.max_depth` (line 68) is also caller-supplied, and `traverse` only checks `len(path) > max_depth`, so the caller sets its own depth cap. `explore` clamps `n` and `max_depth` and `data_preview` clamps `limit`, so these two routes are the gap. Please bound `limit` (for example `Field(100, ge=1, le=1000)`) and require `offset >= 0`. Also enforce a server-side maximum for traverse depth and for the number of `start_ids`. ########## server/ontogeny/functions/sandbox.py: ########## @@ -0,0 +1,320 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Function sandbox: subprocess isolation + capability-scoped query SDK. + +The child never touches the database: every ``ontogeny.query`` travels over stdio +JSON-RPC to this parent, which enforces the declared capabilities +(``read-objects: [...]``) before executing read-only queries. Hard resource +limits (CPU / address space / file descriptors) are applied at spawn time on +POSIX; a wall-clock timeout kills runaway processes. +""" +from __future__ import annotations + +import asyncio +import datetime as _dt +import json +import os +import sys +from pathlib import Path +from typing import Any + +from ..errors import SandboxError +from ..stores.repo import ObjectRepository + + +def _limits() -> None: # pragma: no cover -- preexec in the child process + try: + import resource + + resource.setrlimit(resource.RLIMIT_CPU, (10, 10)) + try: + resource.setrlimit(resource.RLIMIT_AS, (512 << 20, 512 << 20)) + except (ValueError, OSError): + pass # macOS disallows RLIMIT_AS in some configurations + resource.setrlimit(resource.RLIMIT_NOFILE, (64, 64)) + except Exception: + pass + + +def _jsonable(value: Any) -> Any: + if isinstance(value, (_dt.datetime, _dt.date)): + return value.isoformat() + if isinstance(value, dict): + return {k: _jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_jsonable(v) for v in value] + return value + + +def function_source_path(compiled, function_resource) -> Path: + """Where this function's code must live: ``<package_root>/functions/<file>``. + + Resolution only — it does NOT require the file to exist. Callers that need + to run the function (the sandbox) check that themselves and can explain the + failure; callers that need to *report* the situation (the console showing + "what code runs here?") need the path even when it is missing, which is + exactly the case a canvas-created function starts in. + """ + file, _, func = function_resource.spec.entry.partition(":") + if not func: + raise SandboxError(f"entry must be 'file.py:function', got {function_resource.spec.entry!r}") + return Path(compiled.package_root) / "functions" / file + + +def _llm_budget(cfg: dict | None) -> int: + """How many model calls a declaration permits. + + ``max-calls: 0`` means ZERO, not "unset": the declaration is authoritative, + and a falsy-zero bug here would grant MORE calls than declared (10), which is + the one direction a budget must never fail in. + """ + if not cfg or cfg.get("max-calls") is None: + return 10 + return max(0, int(cfg["max-calls"])) + + +def sandbox_env(server_dir: str) -> dict[str, str]: + """Environment allowlist, not inheritance: the child used to receive the + WHOLE process env (DB DSN, HUGEGRAPH credentials, LLM API keys). A + sandboxed function is package code, but "resource fence, not security + boundary" is a claim we only get to make if secrets are not handed over + by default. Operators can widen the list explicitly via + ``ONTOGENY_SANDBOX_ENV_PASSTHROUGH`` (comma-separated names).""" + passthrough = {"PATH", "LANG", "LC_ALL", "LC_CTYPE", "HOME", "TMPDIR", + "SYSTEMROOT", "COMSPEC", "PYTHONIOENCODING", "PYTHONUTF8"} + for var in os.environ.get("ONTOGENY_SANDBOX_ENV_PASSTHROUGH", "").split(","): + if var.strip(): + passthrough.add(var.strip()) + env = {k: v for k, v in os.environ.items() if k in passthrough} + env["PYTHONPATH"] = os.pathsep.join( + p for p in (server_dir, env.get("PYTHONPATH", "")) if p + ) + # never drop __pycache__ into the user's ontology package (it is a git + # working tree: bytecode would show up as untracked noise and would + # change the package's content hash for anything that digests the tree) + env["PYTHONDONTWRITEBYTECODE"] = "1" + return env + + +class FunctionSandbox: + def __init__(self, compiled, repo: ObjectRepository, *, timeout_s: float = 10.0, + llm=None, llm_call_timeout_s: float = 60.0) -> None: + self.compiled = compiled + self.repo = repo + self.timeout_s = timeout_s + self.llm = llm # platform LLM gateway (ontogeny.llm.ChatClient contract); None = not configured + # Wall clock a single ``ontogeny.llm`` round trip may take. The base budget + # above is deliberately tight -- it bounds local work on object rows -- + # but a model round trip is not local work: a 27B model over the network + # routinely answers in 10s+, so a function that declares the capability + # would ALWAYS be killed by a 10s wall clock before its model reply came + # back. See ``_budget`` for how the two combine. + self.llm_call_timeout_s = llm_call_timeout_s + + def _budget(self, function_resource) -> float: + """Wall clock allowed for one invocation of this function. + + The declared capability buys the time it needs, in proportion to its own + declared call budget: a function that may make N model calls gets the + local budget plus N call timeouts. Everything is still bounded -- the + extra time is never open-ended, and a function that declares no ``llm`` + keeps the original tight limit. + """ + cfg = self._llm_capability(function_resource) + if cfg is None: + return self.timeout_s + calls = _llm_budget(cfg) + return self.timeout_s + max(1, calls) * self.llm_call_timeout_s + + def _read_objects_capability(self, function_resource) -> set[str]: + allowed: set[str] = set() + for cap in function_resource.spec.capabilities: + data = cap.model_dump(exclude_none=True) + if "read-objects" in data: + allowed.update(data["read-objects"]) + return allowed + + def _llm_capability(self, function_resource) -> dict | None: + """Declared llm capability config: True -> {} / {model, max-calls, ...}.""" + for cap in function_resource.spec.capabilities: + data = cap.model_dump(exclude_none=True) + if "llm" in data: + cfg = data["llm"] + return cfg if isinstance(cfg, dict) else {} + return None + + def _entry_path(self, function_resource) -> Path: + file, _, func = function_resource.spec.entry.partition(":") + if not func: + raise SandboxError(f"entry must be 'file.py:function', got {function_resource.spec.entry!r}") + root = Path(self.compiled.package_root) + path = function_source_path(self.compiled, function_resource) + if not path.is_file(): + if not root.is_dir() or not (root / "ontology.yaml").is_file(): + raise SandboxError( + f"cannot locate function source {file!r}: the ontology snapshot was " + f"loaded from the registry and has no package directory on disk. " + f"Set ONTOGENY_PACKAGE_ROOT to the checked-out ontology package " + f"(currently {self.compiled.package_root!r}).", + details={"entry": function_resource.spec.entry, "package_root": str(root)}, + ) + raise SandboxError(f"function source not found: {path}") + return path + + async def run(self, session, function_resource, params: dict[str, Any]) -> Any: + return await self._spawn(session, function_resource, + {"type": "invoke", "params": _jsonable(params)}) + + async def run_source(self, session, function_resource, source: str, + params: dict[str, Any]) -> Any: + """One invocation of text straight from the editor, persisting nothing. + + The console's "Test run" needs an answer for code that may never be + saved, so the buffer is written to a throwaway file OUTSIDE the package + (a stray file in ``functions/`` would change the package's content hash + and show up as git noise) and the child loads that instead. Everything + else — entry name, capabilities, budget — is still the declared + function's, so a test run proves exactly what a save would run. + """ + import tempfile + + with tempfile.TemporaryDirectory(prefix="ontogeny-fn-test-") as tmp: + path = Path(tmp) / "buffer.py" + path.write_text(source, encoding="utf-8") + return await self._spawn(session, function_resource, + {"type": "invoke", "params": _jsonable(params)}, + path=path) + + async def map_rows(self, session, function_resource, rows: list[dict[str, Any]], + params: dict[str, Any] | None = None, *, object_param: str = "obj") -> list[Any]: + """Batch mode for the derivation worker: ONE subprocess evaluates the + function per row (the loop runs inside the child), values returned in + row order. Same capability enforcement as run().""" + return await self._spawn(session, function_resource, { + "type": "map", "rows": _jsonable(rows), "params": _jsonable(params or {}), + "object_param": object_param, + }) + + async def _spawn(self, session, function_resource, init_message: dict[str, Any], + *, path: Path | None = None) -> Any: + entry = function_resource.spec.entry + path = path or self._entry_path(function_resource) + _, func = entry.split(":", 1) + allowed = self._read_objects_capability(function_resource) + + server_dir = str(Path(__file__).resolve().parent.parent.parent) + env = sandbox_env(server_dir) + + proc = await asyncio.create_subprocess_exec( + sys.executable, "-m", "ontogeny.functions.child", str(path), func, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + env=env, + preexec_fn=_limits if os.name == "posix" else None, + ) + assert proc.stdin and proc.stdout + + llm_cfg = self._llm_capability(function_resource) + llm_budget = _llm_budget(llm_cfg) + llm_used = 0 + + async def child_main() -> Any: + proc.stdin.write((json.dumps(init_message) + "\n").encode()) + await proc.stdin.drain() + nonlocal llm_used + while True: + line = await proc.stdout.readline() + if not line: + raise SandboxError("sandbox exited before returning a result") + msg = json.loads(line) Review Comment: Important: A sandboxed function that calls `print()` always fails, and the caller gets a raw 500. The child writes its protocol messages to `sys.stdout`, and the user code writes to the same stream, so the printed line reaches this `json.loads` first. `JSONDecodeError` is neither a `SandboxError` nor an `OOError`, so it escapes `wait_for` and the API error handler. Reproduced at this head with `functions/f.py` containing `def hello(name): print("debug:", name); return {"greeting": "hi " + name}`: `FunctionSandbox.run` raises `json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)`. Derivation map mode loses the whole batch the same way. Please keep a private handle to the real stdout in `child.py` for protocol messages and point `sys.stdout` at `sys.stderr` (or a buffer) before the user module is imported. Also treat a non-JSON line here as a `SandboxError`. ########## server/ontogeny/engine/service.py: ########## @@ -0,0 +1,348 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Query service: object assembly (derived properties + masking), link +expansion, aggregation and multi-hop graph traversal (SQL-recursive default; +the HugeGraph projection is an accelerator behind the same API). +""" +from __future__ import annotations + +import time +from typing import Any + +from ..core import expr as _expr +from ..core.models import ForeignKeyJoin +from ..errors import NotFoundError, StoreError +from ..policy.engine import PolicyEngine +from ..stores.repo import ObjectRepository + + +def assemble_row(compiled, object_type: str, props: dict) -> dict: + """Row props -> object props with derived values computed. + + Shared by the query API and the function sandbox: functions read the same + shape of object a human would, so a derived property is never invisible to + business logic (it used to read as None/0 inside the sandbox). + """ + obj = compiled.objects.get(object_type) + if obj is None: + raise NotFoundError(f"unknown object type {object_type!r}") + out = dict(props) + ctx = _expr.ExprContext(target=props) + for prop, pdef in obj.spec.properties.items(): + # only expression deriveds are computed at read time; function deriveds + # are materialized columns maintained by the derivation worker + if pdef.derived_kind() == "expr": + try: + out[prop] = _expr.evaluate(pdef.derived, ctx) Review Comment: Important: A derived property computed from a marked property returns the marked value unmasked. `assemble_row` evaluates `derived` expressions against the raw row. `PolicyEngine.mask` then masks only properties that carry a `marking` themselves, and the validator does not check what a derived expression reads. Reproduced at this head: adding `cost_hint: {type: decimal(14,2), derived: unit_cost * 1}` to `objects/material.yaml` passes `validate()` with no issues. For an anonymous principal, `mask("material", assemble_row(...))` then returns `unit_cost: __masked__` next to `cost_hint: 12.34`. A realistic derived value such as `unit_cost * stock_qty` leaks the same way through `/objects/*/query`, `get`, link expansion, traverse and explore. Please make the validator reject an expression-derived property that reads a marked property (`expr.analyze` gives the references) unless it carries the same marking, or propagate the marking to the derived property at compile time. ########## server/ontogeny/agent/broker.py: ########## @@ -0,0 +1,727 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""AgentBroker: session lifecycle + governed tool execution. + +Every external agent call funnels through :meth:`call_tool`. The method is +the contract: authenticate the session, check the catalog, execute through +ServiceContext, record the step. Writes additionally go through the approval +gate (M2) -- a pending approval row is created and nothing executes until a +human consumes it. +""" +from __future__ import annotations + +import datetime as _dt +import logging +import time +from typing import Any + +from sqlalchemy import delete, or_, select, update + +from ..action.models import OutboxRow # noqa: F401 (re-exported for API layer typing) +from ..errors import NotFoundError, OOError, PolicyDeniedError, StoreError +from ..registry.compiled import CompiledOntology +from .catalog import ToolDef, build_catalog, catalog_hash, effective_tools +from .models import AgentApprovalRow, AgentSessionRow, AgentStepRow, args_digest + +log = logging.getLogger("ontogeny.agent") + + +def _utcnow() -> _dt.datetime: + return _dt.datetime.now(_dt.timezone.utc) + + +def _jsonable(value: Any) -> Any: + """Row objects carry dates/datetimes/Decimals; JSON columns refuse them.""" + if isinstance(value, (_dt.datetime, _dt.date)): + return value.isoformat() + if isinstance(value, dict): + return {k: _jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_jsonable(v) for v in value] + try: + import json as _json + _json.dumps(value) + return value + except (TypeError, ValueError): + return str(value) + + +class SessionBudgetExceeded(OOError): + code = "AGENT_BUDGET_EXCEEDED" + http_status = 429 + + +class ToolNotAllowed(OOError): + code = "AGENT_TOOL_NOT_ALLOWED" + http_status = 403 + + +class AgentServiceError(StoreError): + """Raised for broker-internal problems (bad session state etc.).""" + + +class AgentBroker: + def __init__(self, sc) -> None: + self.sc = sc + + # ------------------------------------------------------------- catalog + + def plugin(self, name: str) -> Any: + if self.sc.compiled is None: + raise StoreError("service not initialized") + plug = self.sc.compiled.agent_plugins.get(name) + if plug is None: + raise NotFoundError(f"unknown agent plugin {name!r}") + return plug + + def list_plugins(self) -> list[dict[str, Any]]: + if self.sc.compiled is None: + return [] + out = [] + for name, plug in self.sc.compiled.agent_plugins.items(): + tools = effective_tools(self.sc.compiled, plug) + out.append({ + "name": name, + "display": plug.metadata.display, + "description": plug.metadata.description, + "transport": plug.spec.transport.kind, + "engine": {"kind": plug.spec.engine.kind, "endpoint": plug.spec.engine.endpoint}, + "principal": plug.spec.principal.as_principal(), + "approval": plug.spec.approval.writes, + "budget": { + "steps": plug.spec.budget.steps, + "wall_ms": plug.spec.budget.wall_ms, + "writes_per_session": plug.spec.budget.writes_per_session, + }, + "tools": {n: t.to_meta() for n, t in tools.items()}, + }) + return out + + # ------------------------------------------------------------ sessions + + async def open_session(self, plugin_name: str, task: str, *, + budget: dict[str, Any] | None = None, + expires_at: _dt.datetime | None = None, + driver: dict[str, Any] | None = None) -> dict[str, Any]: + """Open a session. ``budget`` overrides the plugin's frozen limits per + field; 0 (or a negative) means UNLIMITED, an omitted field keeps the + plugin's value. ``expires_at`` bounds the session's usable life. + ``driver`` is the accountable WHO behind the session (the plugin + principal stays the EXECUTION identity; this records who asked for it). + """ + plug = self.plugin(plugin_name) + compiled: CompiledOntology = self.sc.compiled # type: ignore[assignment] + tools = effective_tools(compiled, plug) + budget = budget or {} + + def _limit(field: str, plugin_default: int) -> int: + v = budget.get(field) + if v is None: + return plugin_default + return max(int(v), 0) # 0 = unlimited + + async with self.sc.sessionmaker() as session: + row = AgentSessionRow( + plugin=plugin_name, + principal=plug.spec.principal.id, + task=task or "", + budget_steps=_limit("steps", plug.spec.budget.steps), + budget_wall_ms=_limit("wall_ms", plug.spec.budget.wall_ms), + budget_writes=_limit("writes_per_session", plug.spec.budget.writes_per_session), + expires_at=expires_at, + catalog_hash=catalog_hash(tools), + driver={"id": driver.get("id", "?"), "via": driver.get("via", "?")} + if driver else None, + run_started_at=_utcnow(), + # freeze the catalog: a plugin edit mid-session can neither + # widen nor narrow a running session (the documented contract) + tools_json={n: t.to_meta() for n, t in tools.items()}, + ) + session.add(row) + await session.commit() + return self._session_view(row, tools) + + async def start_run(self, session_id: int, task: str | None = None) -> dict[str, Any]: + """Direction 3: a session may be driven MANY times. Each run bumps the + run counter (new steps are tagged with it) and re-opens a finished or + budget-exhausted session so the engine can drive it again. ``task`` is + this run's instruction: when given, it becomes the session's task (the + creation-time text is a description, not the command).""" + async with self.sc.sessionmaker() as session: + row = (await session.execute( + select(AgentSessionRow).where(AgentSessionRow.id == session_id) + )).scalar_one_or_none() + if row is None: + raise NotFoundError(f"agent session {session_id} not found") + if self._check_expired(row): + raise OOError("agent session expired", + details={"code": "AGENT_SESSION_EXPIRED"}) + if row.status == "running": + return {"id": row.id, "status": "running", "run_count": row.run_count} + if task and task.strip(): + row.task = task.strip() + row.status = "open" + row.finished_at = None + row.run_count += 1 + # a fresh run gets a fresh wall clock: wall_ms bounds one drive, + # not the session's whole multi-run life + row.run_started_at = _utcnow() + await session.commit() + return {"id": row.id, "status": row.status, "run_count": row.run_count} + + def _check_expired(self, row: AgentSessionRow) -> bool: + """True when the session is past its expires_at (and marks it).""" + # SQLite returns naive UTC datetimes; normalize before comparing + exp = row.expires_at.replace(tzinfo=_dt.timezone.utc) \ + if row.expires_at is not None and row.expires_at.tzinfo is None \ + else row.expires_at + if exp is not None and _utcnow() > exp: + row.status = "expired" + return True + return False + + async def get_session(self, session_id: int) -> dict[str, Any]: + async with self.sc.sessionmaker() as session: + row = (await session.execute( + select(AgentSessionRow).where(AgentSessionRow.id == session_id) + )).scalar_one_or_none() + if row is None: + raise NotFoundError(f"agent session {session_id} not found") + steps = (await session.execute( + select(AgentStepRow).where(AgentStepRow.session_id == session_id) + .order_by(AgentStepRow.seq) + )).scalars().all() + plug = self.sc.compiled.agent_plugins.get(row.plugin) if self.sc.compiled else None + tools = self._frozen_tools(row, plug) + return self._session_view(row, tools, steps=[self._step_view(s) for s in steps]) + + async def list_sessions(self, *, plugin: str | None = None, limit: int = 50) -> list[dict[str, Any]]: + stmt = select(AgentSessionRow).order_by(AgentSessionRow.id.desc()).limit(limit) + if plugin: + stmt = stmt.where(AgentSessionRow.plugin == plugin) + async with self.sc.sessionmaker() as session: + rows = (await session.execute(stmt)).scalars().all() + return [self._session_view(r) for r in rows] + + async def delete_session(self, session_id: int) -> dict[str, Any]: + """Delete one agent session and every observation belonging to it. + + The session row is the governance record's root: steps and approvals + are meaningless without it, so the delete is deliberately transitive. + """ + async with self.sc.sessionmaker() as session: + row = await session.get(AgentSessionRow, session_id) + if row is None: + raise NotFoundError(f"agent session {session_id} not found") + await session.execute(delete(AgentStepRow).where(AgentStepRow.session_id == session_id)) + await session.execute(delete(AgentApprovalRow).where(AgentApprovalRow.session_id == session_id)) Review Comment: Important: Any signed-in driver can erase another session's governance trail, decided approvals included, through `DELETE /api/v1/agent/sessions/{id}`. The route is gated only by `_require_driver`, which passes any authenticated principal (and anyone when dev_auth is on), and it never compares the caller with the session's `driver`. This method then deletes every `AgentStepRow` and every `AgentApprovalRow` of the session, whatever their status. `DELETE /agent/approvals/{id}` is admin-only because deleting a decided approval erases part of the audit trail (`routes_agent.py:233-234`), and this route gets around that gate. An operator whose agent made a write can delete the step that names the run and the approval that names who approved it. Please require admin for session deletion, or limit it to the session's own driver and to sessions with no decided approvals or executed steps. Also keep decided approvals out of the transitive delete. ########## server/ontogeny/auth/service.py: ########## @@ -0,0 +1,270 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Accounts, sessions and the administrator gate. + +One service over two tables (see ``models.py``). It owns: + +- **login/logout** — verify a password, mint a bearer token, keep only its + digest, and resolve a token back into the principal dict the engine consumes; +- **account administration** — create/update/disable/delete users, reset + passwords, grant or revoke roles; +- **bootstrap** — make sure an administrator exists so a fresh deployment is + reachable at all. + +Session lifetime is deliberately short and sliding-free: an expired session is +re-authenticated, not silently extended. +""" +from __future__ import annotations + +import datetime as _dt +import logging +import os +from typing import Any + +from sqlalchemy import delete, func, select + +from ..errors import AuthenticationError, DSLValidationError, NotFoundError +from .models import SessionRow, UserRow +from .passwords import hash_password, new_token, token_fingerprint, verify_password + +log = logging.getLogger("ontogeny.auth") + +SESSION_TTL = _dt.timedelta(hours=12) +MIN_PASSWORD = 8 + + +class AuthService: + def __init__(self, sc) -> None: + self.sc = sc + + # ------------------------------------------------------------------ users + + async def _user(self, session, username: str) -> UserRow | None: + return (await session.execute( + select(UserRow).where(UserRow.username == username) + )).scalar_one_or_none() + + @staticmethod + def _view(row: UserRow) -> dict[str, Any]: + return { + "id": row.id, + "username": row.username, + "display": row.display or row.username, + "roles": list(row.roles or []), + "site": row.site, + "markings": list(row.markings or []), + "status": row.status, + "is_admin": bool(row.is_admin), + "created_at": row.created_at.isoformat() if row.created_at else None, + "last_login_at": row.last_login_at.isoformat() if row.last_login_at else None, + } + + async def list_users(self) -> list[dict[str, Any]]: + async with self.sc.sessionmaker() as session: + rows = (await session.execute(select(UserRow).order_by(UserRow.username))).scalars().all() + return [self._view(r) for r in rows] + + async def create_user(self, *, username: str, password: str, display: str = "", + roles: list[str] | None = None, site: str | None = None, + markings: list[str] | None = None, is_admin: bool = False) -> dict[str, Any]: + username = (username or "").strip() + if not username: + raise DSLValidationError("username is required") + self._check_password(password) + async with self.sc.sessionmaker() as session: + if await self._user(session, username): + raise DSLValidationError(f"user {username!r} already exists") + row = UserRow( + username=username, display=display or username, + password_hash=hash_password(password), + roles=sorted(set(roles or [])), site=site or None, + markings=sorted(set(markings or [])), is_admin=bool(is_admin), + status="active", + ) + session.add(row) + await session.commit() + await session.refresh(row) + return self._view(row) + + async def update_user(self, username: str, patch: dict[str, Any]) -> dict[str, Any]: + async with self.sc.sessionmaker() as session: + row = await self._user(session, username) + if row is None: + raise NotFoundError(f"user {username!r} not found") + if "display" in patch and patch["display"] is not None: + row.display = str(patch["display"]) + if patch.get("roles") is not None: + row.roles = sorted({str(r) for r in patch["roles"]}) + if "site" in patch: + row.site = patch["site"] or None + if patch.get("markings") is not None: + row.markings = sorted({str(m) for m in patch["markings"]}) + if patch.get("status"): + if patch["status"] not in ("active", "disabled"): + raise DSLValidationError("status must be active or disabled") + row.status = patch["status"] + if patch.get("is_admin") is not None: Review Comment: Important: The last administrator can be demoted, and the server then fails to start on its next restart. `update_user` accepts `is_admin: false` or `status: disabled` for any account and never checks that another active administrator remains. `delete_user` refuses to delete administrators to prevent exactly this lockout. After `PATCH /api/v1/admin/users/admin {"is_admin": false}`, the next `initialize()` runs `ensure_admin()`, which finds no administrator and calls `create_user(username="admin")`. That raises `DSLValidationError: user 'admin' already exists`, so startup fails (reproduced at this head with `AuthService` on in-memory SQLite). Disabling the only admin instead leaves it counted by `ensure_admin`, so nobody can sign in as an administrator and the bootstrap never recovers. Please refuse any patch that removes `is_admin` from, or disables, the last active administrator. Also make `ensure_admin` count only active administrators and handle an existing non-admin account wi th the bootstrap username without raising. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
