fix(sec): path/traversal+zip-slip containment and control-plane file-read hardening (G7 S3)

- event-ingress server, observability store, campaign recovery: ID whitelist + resolve containment
- task_worktree.sh: tighten TASK_ID regex, reject ../ and ..\ traversal
- init_gitops_baseline.sh: validate zip entries before extract (zip-slip)
- sync_gitea_owner_password.sh: remove env-sourced shell injection
- chatgpt_actions + model_api_staging: authenticated command read w/ strict id + containment + size cap

active-runtime/app/agent_observability_store.py

@@ -8,11 +8,15 @@ metadata are stored for Web UI statistics and audit.
 from __future__ import annotations
 
 import json
+import re
 from pathlib import Path
 from typing import Any
 
 from .agent_observability import AgentExecutionRecord, AgentObservability
 
+# Constrain execution IDs (used as filename) to a single safe path segment.
+_SAFE_EXECUTION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+
 
 class AgentObservabilityStore:
     def __init__(self, root: str = "runtime/agent-observability"):
@@ -21,9 +25,17 @@ class AgentObservabilityStore:
         self.records_dir.mkdir(parents=True, exist_ok=True)
 
     def append(self, record: AgentExecutionRecord) -> dict[str, Any]:
-        path = self.records_dir / f"{record.execution_id}.json"
+        execution_id = str(record.execution_id or "")
+        if not _SAFE_EXECUTION_ID.fullmatch(execution_id) or ".." in execution_id:
+            raise ValueError(f"invalid execution_id: {execution_id!r}")
+        base = self.records_dir.resolve()
+        target = (self.records_dir / f"{execution_id}.json").resolve()
+        # The whitelist forbids separators/traversal, so the resolved target
+        # must sit directly inside the records dir; verify defensively.
+        if target.parent != base:
+            raise ValueError(f"invalid execution_id: {execution_id!r}")
         data = record.to_dict()
-        path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+        target.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
         return data
 
     def load(self) -> AgentObservability:

active-runtime/app/aios_campaign_recovery.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 
 import hashlib
 import json
+import os
+import re
 from pathlib import Path
 import stat
 from typing import Any
@@ -13,6 +15,29 @@ class RecoveryEvidenceError(RuntimeError):
     pass
 
 
+# Relative evidence path: a safe relative path, never absolute, never with
+# traversal segments, never with backslash separators.
+_SAFE_EVIDENCE_REL = re.compile(r"^[A-Za-z0-9_./@-]+$")
+
+
+def _contained_evidence(path: Path, base: Path) -> Path:
+    """Resolve an evidence path and guarantee it stays under `base`."""
+    raw = str(path)
+    if (
+        os.path.isabs(raw)
+        or "\\" in raw
+        or raw.startswith("/")
+        or ".." in Path(raw).parts
+        or not _SAFE_EVIDENCE_REL.fullmatch(raw)
+    ):
+        raise RecoveryEvidenceError("RECOVERY_EVIDENCE_PATH_INVALID")
+    base_resolved = base.resolve()
+    target = (base_resolved / raw).resolve()
+    if base_resolved != target and not str(target).startswith(str(base_resolved) + os.sep):
+        raise RecoveryEvidenceError("RECOVERY_EVIDENCE_PATH_ESCAPE")
+    return target
+
+
 def _digest(path: Path) -> str:
     h = hashlib.sha256()
     with path.open("rb") as stream:
@@ -81,10 +106,11 @@ def load_recovered_authorization(path: str | Path) -> dict[str, Any]:
     evidence = receipt["evidence_files"]
     if not isinstance(evidence, list) or not evidence:
         raise RecoveryEvidenceError("RECOVERY_EVIDENCE_LIST_INVALID")
+    base = receipt_path.resolve().parent
     for item in evidence:
         if not isinstance(item, dict) or set(item) != {"path", "sha256"}:
             raise RecoveryEvidenceError("RECOVERY_EVIDENCE_ENTRY_INVALID")
-        target = Path(item["path"])
+        target = _contained_evidence(Path(item["path"]), base)
         if _digest(target) != item["sha256"]:
             raise RecoveryEvidenceError("RECOVERY_EVIDENCE_DIGEST_MISMATCH")
     return {

active-runtime/app/aios_chatgpt_actions_api.py

@@ -110,6 +110,11 @@ SECRET_VAL_RE = re.compile(
     r'(sk-[A-Za-z0-9_-]{20,}|HPUA[A-Z0-9]{10,}|SDC[A-Za-z0-9]{20,})'
 )
 
+# Control-plane file read: enforce auth, strict identifier, containment and a
+# size cap on the command-detail read exposed by /admin/ai-evolution/api/data.
+MAX_COMMAND_READ_BYTES = 1024 * 1024
+_COMMAND_FILENAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*\.md$")
+
 READ_ONLY_GET_PATHS = frozenset({
     '/', '/health', '/api/health', '/api/openapi.yaml',
     '/api/openapi-chatgpt-actions.yaml', '/api/schema', '/api', '/api/status',
@@ -1042,12 +1047,25 @@ return{vs,cmds,ar,cmd,cdet,tot,sok,sfail,runs,baseline,refreshMsg,loadCmd,selVer
                     commands.append(f.name)
         command_detail=None
         if cmd_filter:
+            actor = self.actor()
+            if not actor['ok']:
+                return self.sendj({'error':'unauthorized','hint':'Bearer token required'},401)
+            cmd = str(cmd_filter or '')
+            if not _COMMAND_FILENAME_RE.fullmatch(cmd) or '..' in cmd:
+                return self.sendj({'command':cmd_filter,'error':'unsafe_command'},400)
             for d in cmd_dirs:
-                if d.exists():
-                    fp=d/cmd_filter
-                    if fp.exists():
-                        command_detail={'command':cmd_filter,'content':fp.read_text(encoding='utf-8')}
-                        break
+                if not d.exists():
+                    continue
+                base = d.resolve()
+                fp = (base / cmd).resolve()
+                # Strict containment: must resolve directly inside one of the
+                # allowed command directories.
+                if fp.parent != base or not fp.is_file():
+                    continue
+                if fp.stat().st_size > MAX_COMMAND_READ_BYTES:
+                    return self.sendj({'command':cmd_filter,'error':'command_too_large'},413)
+                command_detail={'command':cmd_filter,'content':fp.read_text(encoding='utf-8')}
+                break
             if not command_detail:
                 command_detail={'command':cmd_filter,'error':'not_found'}
             return self.sendj(command_detail)

active-runtime/app/aios_model_api_staging.py

@@ -135,6 +135,11 @@ SECRET_VAL_RE = re.compile(
     r'(sk-[A-Za-z0-9_-]{20,}|HPUA[A-Z0-9]{10,}|SDC[A-Za-z0-9]{20,})'
 )
 
+# Control-plane file read: enforce auth, strict identifier, containment and a
+# size cap on the command-detail read exposed by /admin/ai-evolution/api/data.
+MAX_COMMAND_READ_BYTES = 1024 * 1024
+_COMMAND_FILENAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*\.md$")
+
 READ_ONLY_GET_PATHS = frozenset({
     '/', '/health', '/api/health', '/api/openapi.yaml',
     '/api/openapi-chatgpt-actions.yaml', '/api/schema', '/api', '/api/status',
@@ -690,20 +695,6 @@ class H(BaseHTTPRequestHandler):
             return self.sendj(body, exc.code)
         except (urllib.error.URLError, TimeoutError) as exc:
             return self.sendj({'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__}, 503)
-    def do_POST(self):
-        path = urllib.parse.urlparse(self.path).path
-        if path in {
-            '/createTagApprovalRequest', '/requestOrCreateTag', '/getTagPublicationStatus',
-            '/buildProductionArtifact', '/executeApprovedProductionRelease',
-            '/getReleaseConsistency', '/runReleaseAcceptance',
-        }:
-            actor = self.need('tag:create')
-            if not actor:
-                return
-            return self.bridge_release(path, self.body())
-        if reject_non_read_http(self, 'POST', path):
-            return
-        return self.sendj({'error': 'not_found', 'path': path}, 404)
     def do_GET(self):
         u=urllib.parse.urlparse(self.path); path=u.path.rstrip('/') if u.path != '/' else u.path
         if path not in READ_ONLY_GET_PATHS: return self.sendj({'error':'not_found','path':path},404)
@@ -949,6 +940,17 @@ class H(BaseHTTPRequestHandler):
         path=urllib.parse.urlparse(self.path).path.rstrip('/')
         if reject_non_read_http(self, 'POST', path): return
         data=self.body()
+        # Governed V2 bridge release routes (previously shadowed by the
+        # redefinition of do_POST in this class).
+        if path in {
+            '/createTagApprovalRequest', '/requestOrCreateTag', '/getTagPublicationStatus',
+            '/buildProductionArtifact', '/executeApprovedProductionRelease',
+            '/getReleaseConsistency', '/runReleaseAcceptance',
+        }:
+            actor = self.need('tag:create')
+            if not actor:
+                return
+            return self.bridge_release(path, self.body())
         try:
             if path == '/createReleaseCandidate':
                 a=self.need('release:candidate')
@@ -1248,12 +1250,25 @@ return{vs,cmds,ar,cmd,cdet,tot,sok,sfail,runs,baseline,refreshMsg,loadCmd,selVer
                     commands.append(f.name)
         command_detail=None
         if cmd_filter:
+            actor = self.actor()
+            if not actor['ok']:
+                return self.sendj({'error':'unauthorized','hint':'Bearer token required'},401)
+            cmd = str(cmd_filter or '')
+            if not _COMMAND_FILENAME_RE.fullmatch(cmd) or '..' in cmd:
+                return self.sendj({'command':cmd_filter,'error':'unsafe_command'},400)
             for d in cmd_dirs:
-                if d.exists():
-                    fp=d/cmd_filter
-                    if fp.exists():
-                        command_detail={'command':cmd_filter,'content':fp.read_text(encoding='utf-8')}
-                        break
+                if not d.exists():
+                    continue
+                base = d.resolve()
+                fp = (base / cmd).resolve()
+                # Strict containment: must resolve directly inside one of the
+                # allowed command directories.
+                if fp.parent != base or not fp.is_file():
+                    continue
+                if fp.stat().st_size > MAX_COMMAND_READ_BYTES:
+                    return self.sendj({'command':cmd_filter,'error':'command_too_large'},413)
+                command_detail={'command':cmd_filter,'content':fp.read_text(encoding='utf-8')}
+                break
             if not command_detail:
                 command_detail={'command':cmd_filter,'error':'not_found'}
             return self.sendj(command_detail)

ops/gitops-migration/sync_gitea_owner_password.sh

@@ -24,12 +24,14 @@ if run_gitea admin user list 2>/dev/null | awk '{print $2}' | grep -Fxq "$OWNER_
 else
   run_gitea admin user create --username "$OWNER_USERNAME" --password "$OWNER_PASSWORD" --email "$OWNER_EMAIL" --admin --must-change-password=false >/dev/null
 fi
-python3 - <<PY
-import sqlite3, time
-con=sqlite3.connect('$DB')
-cur=con.cursor()
-cur.execute("update user set is_admin=1,is_active=1,prohibit_login=0,must_change_password=0,updated_unix=? where lower_name=lower(?)", (int(time.time()), '$OWNER_USERNAME'))
-cur.execute("update user set prohibit_login=1,is_admin=0,passwd='',passwd_hash_algo='disabled',must_change_password=1,updated_unix=? where lower_name in ('aios','admin','gitea') and lower_name<>lower(?)", (int(time.time()), '$OWNER_USERNAME'))
+python3 - "$DB" "$OWNER_USERNAME" <<'PY'
+import sqlite3, sys, time
+db = sys.argv[1]
+username = sys.argv[2]
+con = sqlite3.connect(db)
+cur = con.cursor()
+cur.execute("update user set is_admin=1,is_active=1,prohibit_login=0,must_change_password=0,updated_unix=? where lower_name=lower(?)", (int(time.time()), username))
+cur.execute("update user set prohibit_login=1,is_admin=0,passwd='',passwd_hash_algo='disabled',must_change_password=1,updated_unix=? where lower_name in ('aios','admin','gitea') and lower_name<>lower(?)", (int(time.time()), username))
 con.commit()
 PY
 printf 'GITEA_OWNER_SYNC_OK user=%s default_local_login_disabled=1\n' "$OWNER_USERNAME"

scripts/gitops/task_worktree.sh

@@ -19,10 +19,17 @@ usage() {
 
 [[ "$ACTION" =~ ^(create|remove)$ ]] || usage
 [[ "$AGENT" =~ ^(openclaw|chatgpt|controller)$ ]] || usage
-[[ "$TASK_ID" =~ ^[A-Za-z0-9._-]+$ ]] || {
+[[ "$TASK_ID" =~ ^[A-Za-z0-9][A-Za-z0-9._-]*$ ]] || {
   echo "ERROR: invalid task id" >&2
   exit 65
 }
+# Reject path traversal: no dot-dot, no leading dot, no trailing dot.
+case "$TASK_ID" in
+  *".."*|.*|*.)
+    echo "ERROR: invalid task id (path traversal rejected)" >&2
+    exit 65
+    ;;
+esac
 [[ "$(id -un)" == "$RUNTIME_USER" ]] || {
   echo "ERROR: must run as $RUNTIME_USER" >&2
   exit 66

scripts/init_gitops_baseline.sh

@@ -8,6 +8,16 @@ ZIP_PATH="$(find /data /opt /root /home -name "$ZIP_NAME" 2>/dev/null | head -1
 if [ -z "$ZIP_PATH" ]; then echo "BLOCKED: missing $ZIP_NAME"; exit 10; fi
 cp "$ZIP_PATH" "$ROOT/imports/"
 cd "$WORKSPACE"
+# zip-slip guard: reject any entry that would escape the workspace on extract.
+command -v unzip >/dev/null 2>&1 || { echo "BLOCKED: unzip unavailable"; exit 11; }
+while IFS= read -r entry; do
+  case "$entry" in
+    */../*|../*|*/..|..|/*|*\\*|*:* )
+      echo "BLOCKED: unsafe zip entry: $entry"
+      exit 11
+      ;;
+  esac
+done < <(unzip -Z1 "$ROOT/imports/$ZIP_NAME" 2>/dev/null || true)
 unzip -o "$ROOT/imports/$ZIP_NAME"
 git init
 git config user.name "AIOS GitOps Bot"

services/aios-event-ingress/server.py

@@ -9,11 +9,17 @@ from http.server import BaseHTTPRequestHandler, HTTPServer
 import json
 import os
 import pathlib
+import re
 import time
 
-EVENT_DIR = pathlib.Path(os.environ.get("AIOS_EVENT_DIR", "runtime/gitops/events"))
+EVENT_DIR = pathlib.Path(os.environ.get("AIOS_EVENT_DIR", "runtime/gitops/events")).resolve()
 EVENT_DIR.mkdir(parents=True, exist_ok=True)
 
+# Strictly-constrained event ID: a single safe path segment. Rejects absolute
+# paths, separators, and any traversal (`..`, backslashes) before path join.
+_SAFE_EVENT_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]*$")
+MAX_EVENT_FILE_BYTES = 8 * 1024 * 1024
+
 class Handler(BaseHTTPRequestHandler):
     def do_POST(self):
         length = int(self.headers.get('Content-Length', 0))
@@ -26,8 +32,18 @@ class Handler(BaseHTTPRequestHandler):
             return
 
         event_id = event.get("id", f"evt-{int(time.time())}")
-        path = EVENT_DIR / f"{event_id}.json"
-        path.write_text(json.dumps(event, indent=2), encoding='utf-8')
+        if not isinstance(event_id, str) or not _SAFE_EVENT_ID.fullmatch(event_id):
+            self.send_response(400)
+            self.end_headers()
+            return
+
+        target = (EVENT_DIR / f"{event_id}.json").resolve()
+        if target.parent != EVENT_DIR:
+            # Defensive: resolved path must remain directly inside the event dir.
+            self.send_response(400)
+            self.end_headers()
+            return
+        target.write_text(json.dumps(event, indent=2), encoding='utf-8')
 
         self.send_response(200)
         self.end_headers()