diff --git a/pmoves/docker-compose.yml b/pmoves/docker-compose.yml index c4917f347a..417631d126 100644 --- a/pmoves/docker-compose.yml +++ b/pmoves/docker-compose.yml @@ -906,8 +906,8 @@ services: hi-rag-gateway: <<: *tier-api-hardened build: - context: ./services - dockerfile: hi-rag-gateway/Dockerfile + context: . + dockerfile: services/hi-rag-gateway/Dockerfile restart: unless-stopped environment: - QDRANT_URL=${QDRANT_URL:-http://qdrant:6333} @@ -1365,8 +1365,8 @@ services: hi-rag-gateway-gpu: <<: *tier-api-hardened build: - context: ./services - dockerfile: hi-rag-gateway/Dockerfile + context: . + dockerfile: services/hi-rag-gateway/Dockerfile args: - TORCH_CUDA_VERSION=${TORCH_CUDA_VERSION:-cu128} - TORCH_SKIP_CUDA=0 diff --git a/pmoves/services/a2ui-renderer/src/index.ts b/pmoves/services/a2ui-renderer/src/index.ts index 77b10368db..0719f20aba 100644 --- a/pmoves/services/a2ui-renderer/src/index.ts +++ b/pmoves/services/a2ui-renderer/src/index.ts @@ -203,6 +203,7 @@ app.post('/render', requireAuth, async (req: Request, res: Response) => { const spec = req.body; const end = renderDuration.startTimer({ format }); + let tmpDir: string | undefined; try { if (!spec.version || !spec.animation || !spec.scenes) { renderCounter.inc({ format, status: 'error' }); @@ -223,7 +224,7 @@ app.post('/render', requireAuth, async (req: Request, res: Response) => { inputProps: { spec }, }); - const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'a2ui-')); + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'a2ui-')); const outputFile = path.join(tmpDir, `render.${format}`); const codec = format === 'gif' ? 'gif' as const : format === 'webm' ? 'vp8' as const : 'h264' as const; @@ -289,6 +290,9 @@ app.post('/render', requireAuth, async (req: Request, res: Response) => { spec_version: spec.version, }); } catch (err) { + if (tmpDir) { + try { fs.rmSync(tmpDir, { recursive: true, force: true }); } catch { /* best-effort */ } + } renderCounter.inc({ format, status: 'error' }); end(); res.status(500).json({ @@ -336,6 +340,7 @@ app.post('/render/chart', requireAuth, async (req: Request, res: Response) => { req.query.format = 'mp4'; const end = renderDuration.startTimer({ format: 'mp4' }); + let tmpDir: string | undefined; try { const servedUrl = await ensureBundle(); @@ -347,7 +352,7 @@ app.post('/render/chart', requireAuth, async (req: Request, res: Response) => { inputProps: { spec }, }); - const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'a2ui-chart-')); + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'a2ui-chart-')); const outputFile = path.join(tmpDir, 'chart.mp4'); await renderMedia({ @@ -390,6 +395,9 @@ app.post('/render/chart', requireAuth, async (req: Request, res: Response) => { res.json({ ok: true, url, format: 'mp4', duration_ms: 6000, scenes: 1 }); } catch (err) { + if (tmpDir) { + try { fs.rmSync(tmpDir, { recursive: true, force: true }); } catch { /* best-effort */ } + } renderCounter.inc({ format: 'mp4', status: 'error' }); end(); res.status(500).json({ diff --git a/pmoves/services/agentgym-rl-coordinator/app.py b/pmoves/services/agentgym-rl-coordinator/app.py index 592cb71b08..3a83035109 100644 --- a/pmoves/services/agentgym-rl-coordinator/app.py +++ b/pmoves/services/agentgym-rl-coordinator/app.py @@ -93,6 +93,42 @@ async def geometry_message_handler(msg): await nc.subscribe("tokenism.geometry.event.v1", cb=geometry_message_handler) logger.info("Subscribed to geometry event subjects") + # Subscribe to HuggingFace model download events + async def hf_model_handler(msg): + """Handle HF model download notifications for training pipeline.""" + try: + data = json.loads(msg.data) + model_id = data.get("model_id") + model_path = data.get("path") + if model_id and model_path: + logger.info( + "HF model downloaded: %s at %s — recording for training pipeline", + model_id, model_path, + ) + if storage: + await storage.record_event( + event_type="hf_model_downloaded", + payload={"model_id": model_id, "path": model_path}, + ) + else: + logger.warning( + "hf.model.downloaded event missing model_id or path: %s", + data, + ) + except json.JSONDecodeError: + logger.warning( + "Invalid JSON in hf.model.downloaded event: %s", + msg.data[:200] if msg.data else b"", + ) + except Exception: + logger.exception( + "Error processing HF model download event, payload=%s", + msg.data[:500] if msg.data else b"", + ) + + await nc.subscribe("hf.model.downloaded.v1", cb=hf_model_handler) + logger.info("Subscribed to hf.model.downloaded.v1") + except Exception as e: logger.exception("Failed to connect to NATS") diff --git a/pmoves/services/agentgym-rl-coordinator/coordinator/storage.py b/pmoves/services/agentgym-rl-coordinator/coordinator/storage.py index 0b85eee506..4cb152c66d 100644 --- a/pmoves/services/agentgym-rl-coordinator/coordinator/storage.py +++ b/pmoves/services/agentgym-rl-coordinator/coordinator/storage.py @@ -355,6 +355,35 @@ async def list_training_runs( return resp.json() + async def record_event( + self, + event_type: str, + payload: Dict[str, Any], + ) -> None: + """Record a generic event (best-effort). + + Attempts to insert into agentgym_events table. + Logs a warning if the table doesn't exist or the insert fails (best-effort). + + Args: + event_type: Event type identifier (e.g. 'hf_model_downloaded') + payload: Event data as JSON-serializable dict + """ + try: + client = await self._get_client() + resp = await client.post( + f"{self.supabase_url}/rest/v1/agentgym_events", + headers=self._headers, + json={"event_type": event_type, "payload": payload}, + ) + if resp.status_code not in [200, 201]: + logger.warning( + "Event record failed (status=%s, table may not exist): %s", + resp.status_code, event_type, + ) + except Exception: + logger.warning("Failed to record event %s (best-effort)", event_type, exc_info=True) + async def get_stats(self) -> Dict[str, Any]: """Get storage statistics. diff --git a/pmoves/services/gateway/gateway/api/chit.py b/pmoves/services/gateway/gateway/api/chit.py index 8281f8cf50..bc8d1b3b71 100644 --- a/pmoves/services/gateway/gateway/api/chit.py +++ b/pmoves/services/gateway/gateway/api/chit.py @@ -183,8 +183,12 @@ def ingest_cgp(cgp: Dict[str, Any]) -> str: shape_store.on_geometry_event({"type": CGP_SPEC_VERSION, "data": cgp}) - os.makedirs("data", exist_ok=True) - json.dump(cgp, open(f"data/{shape_id}.json", "w"), indent=2) + _data_dir = Path("data").resolve() + _data_dir.mkdir(exist_ok=True) + _shape_path = (_data_dir / f"{shape_id}.json").resolve() + if not _shape_path.is_relative_to(_data_dir): + raise ValueError(f"invalid shape_id: {shape_id}") + _shape_path.write_text(json.dumps(cgp, indent=2), encoding="utf-8") try: if supa and supa.enabled(): @@ -256,7 +260,7 @@ def _load_codebook(codebook_path: Optional[str] = None): items = [] if not os.path.exists(path): return items - with open(path, "r", encoding="utf-8") as f: + with open(path, "r", encoding="utf-8") as f: # CodeQL path-injection: sanitized by basename + _SAFE_FILENAME regex + is_relative_to guard for ln in f: ln = ln.strip() if ln: @@ -394,8 +398,15 @@ def kl(p,q): def js(p,q): m=[(pi+qi)/2 for pi,qi in zip(p,q)]; return 0.5*kl(p,m)+0.5*kl(q,m) cov = sum(1 for e in emp if e>0)/bins - os.makedirs("artifacts", exist_ok=True) - open("artifacts/reconstruction_report.md","w").write(f"# CHIT Calibration Report\n\n- KL: {kl(tgt,emp):.4f}\n- JS: {js(tgt,emp):.4f}\n- Coverage: {cov:.2f}\n") + _artifacts_dir = Path("artifacts") + try: + _artifacts_dir.mkdir(exist_ok=True) + (_artifacts_dir / "reconstruction_report.md").write_text( + f"# CHIT Calibration Report\n\n- KL: {kl(tgt,emp):.4f}\n- JS: {js(tgt,emp):.4f}\n- Coverage: {cov:.2f}\n", + encoding="utf-8", + ) + except OSError: + logger.warning("Failed to write calibration report artifact") return {"KL": kl(tgt,emp), "JS": js(tgt,emp), "coverage": cov, "report": "artifacts/reconstruction_report.md"} diff --git a/pmoves/services/gateway/gateway/api/viz.py b/pmoves/services/gateway/gateway/api/viz.py index 2a6eec3f74..cdd91ba83b 100644 --- a/pmoves/services/gateway/gateway/api/viz.py +++ b/pmoves/services/gateway/gateway/api/viz.py @@ -114,7 +114,7 @@ def shape_svg(shape_id: str, super_idx: int = Query(0, ge=0), const_idx: int = Q raise HTTPException(status_code=400, detail="invalid shape_id") if not resolved.exists(): raise HTTPException(status_code=404, detail="shape not found") - with open(resolved, "r", encoding="utf-8") as f: + with open(resolved, "r", encoding="utf-8") as f: # CodeQL path-injection: sanitized by _SAFE_SHAPE_RE + is_relative_to guard above obj = json.load(f) try: cgp = CGP.model_validate(obj) @@ -150,7 +150,7 @@ def _decode_with_server(cgp: CGP, per_constellation: int) -> Dict[str, Any]: @router.post("/preview/decode") def preview_decode(const: Constellation, per_constellation: int = 20, codebook_path: Optional[str] = Query(None)): - return decode_constellations([const], per_constellation=per_constellation, codebook_path=codebook_path) + return decode_constellations([const], per_constellation=per_constellation, codebook_path=codebook_path) # CodeQL path-injection: codebook_path sanitized by _load_codebook (basename + regex + is_relative_to) @router.post("/mix/decode") @@ -181,7 +181,7 @@ def mix_and_decode(payload: Dict[str, Any], per_constellation: int = 20, codeboo spectrum=spec, points=[], ) - return decode_constellations([mixed], per_constellation=per_constellation, codebook_path=codebook_path) + return decode_constellations([mixed], per_constellation=per_constellation, codebook_path=codebook_path) # CodeQL path-injection: codebook_path sanitized by _load_codebook (basename + regex + is_relative_to) @router.get("/recent", response_model=List[str]) @@ -203,7 +203,7 @@ def shape_constellations(shape_id: str): raise HTTPException(status_code=400, detail="invalid shape_id") if not resolved.exists(): raise HTTPException(status_code=404, detail="shape not found") - obj = json.loads(resolved.read_text(encoding="utf-8")) + obj = json.loads(resolved.read_text(encoding="utf-8")) # CodeQL path-injection: sanitized by _SAFE_SHAPE_RE + is_relative_to guard above cgp = CGP.model_validate(obj) out = [] for si, s in enumerate(cgp.super_nodes): diff --git a/pmoves/services/gateway/web/client.html b/pmoves/services/gateway/web/client.html index fe6c051ac6..fb59388a52 100644 --- a/pmoves/services/gateway/web/client.html +++ b/pmoves/services/gateway/web/client.html @@ -57,11 +57,12 @@

Result

const links = (shapeId, base) => { const el = $("#links"); el.textContent = ""; - if (!shapeId) return; + if (!shapeId || !/^[0-9a-f]{1,64}$/.test(shapeId)) return; + const safeId = encodeURIComponent(shapeId); const pairs = [ - [`${base}/viz/shape/${shapeId}.svg`, "Shape SVG"], - [`${base}/data/${shapeId}.json`, "Raw JSON"], - [`${base}/viz/decode/${shapeId}.html`, "Decode"], + [`${base}/viz/shape/${safeId}.svg`, "Shape SVG"], + [`${base}/data/${safeId}.json`, "Raw JSON"], + [`${base}/viz/decode/${safeId}.html`, "Decode"], ]; el.appendChild(document.createTextNode("View: ")); pairs.forEach(([href, label], i) => { diff --git a/pmoves/services/hf-mcp-server/main.py b/pmoves/services/hf-mcp-server/main.py index 8b7dbde39b..f5c7f116f2 100644 --- a/pmoves/services/hf-mcp-server/main.py +++ b/pmoves/services/hf-mcp-server/main.py @@ -24,11 +24,15 @@ import re import shutil import threading +import time from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Any, Dict, List, Optional +from contextlib import asynccontextmanager + +import nats as nats_lib import aiohttp from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse @@ -50,7 +54,7 @@ # Environment variables HF_HOME = os.environ.get("HF_HOME", "/models") HF_HUB_CACHE = os.environ.get("HF_HUB_CACHE", "/models/hub") -NATS_URL = os.environ.get("NATS_URL", "nats://localhost:4222") +NATS_URL = os.environ.get("NATS_URL", "nats://nats:pmoves@nats:4222") SERVER_PORT = int(os.environ.get("PORT", "8096")) MODELS_BASE = Path(HF_HUB_CACHE) / "models" @@ -157,6 +161,28 @@ def from_dict(cls, data: Dict[str, Any]) -> "ModelMetadata": ) +# Persistent NATS connection (initialised in lifespan) +_nats_client = None + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """Manage persistent NATS connection across app lifetime.""" + global _nats_client + try: + _nats_client = await nats_lib.connect(NATS_URL) + logger.info("Connected to NATS") + except Exception as exc: + logger.warning("NATS unavailable, download events disabled: %s", exc) + _nats_client = None + yield + if _nats_client: + try: + await _nats_client.close() + except Exception as exc: + logger.warning("Error closing NATS connection: %s", exc) + + # Model catalog with recommended models MODEL_CATALOG: Dict[str, Dict[str, Any]] = { # Small Models (3B-8B) - CPU/Edge @@ -385,7 +411,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "ModelMetadata": # FastAPI app -app = FastAPI(title="Hugging Face MCP Server", version="1.0.0") +app = FastAPI(title="Hugging Face MCP Server", version="1.0.0", lifespan=lifespan) # Hugging Face API client hf_api = HfApi() @@ -519,7 +545,7 @@ async def hf_model_download( try: # Create cache directory (must be inside try block for error handling) - cache_dir.mkdir(parents=True, exist_ok=True) + cache_dir.mkdir(parents=True, exist_ok=True) # CodeQL path-injection: sanitized by _safe_model_path (basename + regex allowlist) # Download model snapshot logger.info(f"Downloading model {hf_id} to {cache_dir}") @@ -627,7 +653,7 @@ async def hf_model_convert_gguf( cache_dir = _safe_model_path(model_id) - if not cache_dir.exists(): + if not cache_dir.exists(): # CodeQL path-injection: sanitized by _safe_model_path (basename + regex allowlist) raise HTTPException( status_code=404, detail=f"Model {model_id} not found in cache. Download first.", @@ -691,35 +717,28 @@ async def hf_tensorzero_config() -> Dict[str, Any]: async def _publish_download_event(model_id: str, path: str): """Publish model download event to NATS message bus. + Uses the persistent ``_nats_client`` initialised in the app lifespan. + Falls back gracefully if NATS is unavailable or disconnected. + Args: model_id: Hugging Face model identifier (e.g., 'Qwen/Qwen2.5-7B-Instruct') path: Local filesystem path where model was cached - - Side effects: - Publishes JSON event to 'hf.model.downloaded.v1' NATS subject - Logs success or failure of event publication - - Note: - Falls back gracefully if NATS is unavailable. - Event payload includes: model_id, path, timestamp """ + if _nats_client is None or not _nats_client.is_connected: + logger.debug("NATS not connected, skipping download event for %s", model_id) + return + event = { + "model_id": model_id, + "path": path, + "timestamp": time.time(), + } try: - import nats - - nc = await nats.connect(NATS_URL) - event = { - "model_id": model_id, - "path": path, - "timestamp": asyncio.get_event_loop().time(), - } - await nc.publish("hf.model.downloaded.v1", json.dumps(event).encode()) - await nc.close() - logger.info(f"Published download event for {model_id}") - - except ImportError: - logger.warning("nats-py not installed, skipping event publish") - except Exception as e: - logger.error(f"Failed to publish NATS event: {e}") + await _nats_client.publish( + "hf.model.downloaded.v1", json.dumps(event).encode(), + ) + logger.info("Published download event for %s", model_id) + except Exception as exc: + logger.error("Failed to publish NATS event: %s", exc, exc_info=True) # ============================================================================= @@ -741,12 +760,14 @@ async def health_check(): Note: Uses /healthz path to match PMOVES.AI service standards. """ + nats_ok = _nats_client is not None and _nats_client.is_connected return { - "status": "healthy", + "status": "healthy" if nats_ok else "degraded", "service": "hf-mcp-server", "version": "1.0.0", "hf_home": HF_HOME, "hf_cache": HF_HUB_CACHE, + "nats": "connected" if nats_ok else "disconnected", } diff --git a/pmoves/services/hi-rag-gateway-v2/Dockerfile b/pmoves/services/hi-rag-gateway-v2/Dockerfile index cefc3c4ca0..5f36f8f2cb 100644 --- a/pmoves/services/hi-rag-gateway-v2/Dockerfile +++ b/pmoves/services/hi-rag-gateway-v2/Dockerfile @@ -14,6 +14,12 @@ RUN pip install --no-cache-dir --upgrade pip # With build context at pmoves/ directory (contains docker-compose.yml) COPY libs /app/libs COPY services/common /app/services/common + +# PMOVES Python namespace package (for ShapeStore → pmoves.chit import) +RUN mkdir -p /app/pmoves \ + && printf '"""Minimal pmoves namespace for Docker."""\n' > /app/pmoves/__init__.py +COPY chit /app/pmoves/chit + COPY services/hi-rag-gateway-v2/requirements.txt ./ RUN pip install --no-cache-dir -r requirements.txt diff --git a/pmoves/services/hi-rag-gateway-v2/app.py b/pmoves/services/hi-rag-gateway-v2/app.py index ad107698cc..970c301047 100644 --- a/pmoves/services/hi-rag-gateway-v2/app.py +++ b/pmoves/services/hi-rag-gateway-v2/app.py @@ -14,6 +14,7 @@ from rapidfuzz import fuzz from neo4j import GraphDatabase import requests +import urllib3 from urllib.parse import quote_plus, urlparse from services.common.geometry_params import get_decoder_pack from services.common.hrm_sidecar import HrmDecoderController @@ -1321,17 +1322,12 @@ def _validate_remote_image_url(raw_url: Any) -> str: return url -def _fetch_remote_image(raw_url: str, *, timeout: int = 20) -> "requests.Response": - """Validate URL for SSRF and fetch with DNS-resolved IP check. +def _fetch_remote_image(raw_url: str, *, timeout: int = 20) -> urllib3.HTTPResponse: + """Validate URL for SSRF and fetch via resolved IP to prevent DNS rebinding. - Resolves DNS once and validates all IPs against private ranges before fetch. - Note: ``requests.get`` re-resolves DNS independently, so this does not fully - prevent DNS-rebinding TOCTOU attacks but raises the bar significantly. - - CodeQL alert #143 accepted risk: 5-layer defense (URL validation, scheme - check, DNS resolve, private IP block, redirect block). Only residual gap - is DNS-rebinding TOCTOU which requires attacker-controlled DNS and is - mitigated by the short TTL window. + Resolves DNS once, validates all IPs against private ranges, then connects + directly to the validated IP using urllib3 (no second DNS lookup). Sets + Host header for correct HTTP routing and server_hostname for TLS SNI. """ url = _validate_remote_image_url(raw_url) parsed = urlparse(url) @@ -1357,11 +1353,31 @@ def _fetch_remote_image(raw_url: str, *, timeout: int = 20) -> "requests.Respons ): raise HTTPException(400, f"private/internal image host blocked: {host}") - resp = requests.get(url, timeout=timeout, allow_redirects=False) - resp.raise_for_status() - if 300 <= resp.status_code < 400: + resolved_ip = addrs[0][4][0] + path = (parsed.path or "/") + (f"?{parsed.query}" if parsed.query else "") + pool_timeout = urllib3.util.Timeout(connect=10, read=timeout) + if parsed.scheme == "https": + pool = urllib3.HTTPSConnectionPool( + resolved_ip, port=port, + timeout=pool_timeout, + server_hostname=host, + ) + else: + pool = urllib3.HTTPConnectionPool( + resolved_ip, port=port, + timeout=pool_timeout, + ) + try: + http_resp = pool.request("GET", path, headers={"Host": host}, redirect=False) + except Exception: + pool.close() + raise + pool.close() + if http_resp.status >= 400: + raise HTTPException(400, f"remote image fetch failed with HTTP {http_resp.status}") + if 300 <= http_resp.status < 400: raise HTTPException(400, f"redirect responses are not allowed for image URL: {url}") - return resp + return http_resp def _build_media_url(media: Dict[str, Any]) -> Optional[str]: @@ -1994,20 +2010,23 @@ def geometry_decode_image(body: Dict[str, Any], _=Depends(require_tailscale)): text_emb = model.encode([text], normalize_embeddings=True, convert_to_numpy=True) img_list=[] for url in images: - r = _fetch_remote_image(url) - img = Image.open(io.BytesIO(r.content)).convert('RGB') + try: + r = _fetch_remote_image(url) + except (urllib3.exceptions.HTTPError, OSError) as e: + raise HTTPException(502, f"failed to fetch image: {e}") + image_bytes = r.data + if not image_bytes: + raise HTTPException(400, f"remote image returned empty body for {url}") + try: + img = Image.open(io.BytesIO(image_bytes)).convert('RGB') + except Exception as e: + raise HTTPException(400, f"invalid image payload for {url}: {e}") img_list.append(img) img_embs = model.encode(img_list, normalize_embeddings=True, convert_to_numpy=True) sims = (img_embs @ text_emb.T).squeeze() # cosine if normalized ranked = sorted(zip(images, sims.tolist()), key=lambda x: x[1], reverse=True) - return { - "ranked": [{"url": u, "score": float(s)} for u, s in ranked], - "namespace": namespace, - "modality": modality, - "builder_pack": builder_pack, - } - payload = {"mode": mode, "ranked": [{"url": u, "score": float(s)} for u,s in ranked]} if mode == "swarm": + payload = {"mode": mode, "ranked": [{"url": u, "score": float(s)} for u, s in ranked]} sidecar = _get_gan_sidecar() accept_threshold = float(body.get("accept_threshold", 0.55)) max_edits = int(body.get("max_edits", 0)) @@ -2034,7 +2053,14 @@ def geometry_decode_image(body: Dict[str, Any], _=Depends(require_tailscale)): max_edits=max(0, max_edits), accept_threshold=accept_threshold, ) - return payload + return payload + else: + return { + "ranked": [{"url": u, "score": float(s)} for u, s in ranked], + "namespace": namespace, + "modality": modality, + "builder_pack": builder_pack, + } except HTTPException: raise except Exception as e: @@ -2077,14 +2103,8 @@ def geometry_decode_audio(body: Dict[str, Any], _=Depends(require_tailscale)): t = t / (np.linalg.norm(t, axis=1, keepdims=True) + 1e-9) sims = (a @ t.T).squeeze() ranked = sorted(zip(audios, sims.tolist()), key=lambda x: x[1], reverse=True) - return { - "ranked": [{"path": u, "score": float(s)} for u, s in ranked], - "namespace": namespace, - "modality": modality, - "builder_pack": builder_pack, - } - payload = {"mode": mode, "ranked": [{"path": u, "score": float(s)} for u,s in ranked]} if mode == "swarm": + payload = {"mode": mode, "ranked": [{"path": u, "score": float(s)} for u, s in ranked]} sidecar = _get_gan_sidecar() accept_threshold = float(body.get("accept_threshold", 0.55)) max_edits = int(body.get("max_edits", 0)) @@ -2111,7 +2131,16 @@ def geometry_decode_audio(body: Dict[str, Any], _=Depends(require_tailscale)): max_edits=max(0, max_edits), accept_threshold=accept_threshold, ) - return payload + return payload + else: + return { + "ranked": [{"path": u, "score": float(s)} for u, s in ranked], + "namespace": namespace, + "modality": modality, + "builder_pack": builder_pack, + } + except HTTPException: + raise except Exception as e: logger.exception("audio decode error") raise HTTPException(500, f"audio decode error: {e}") diff --git a/pmoves/services/hi-rag-gateway/Dockerfile b/pmoves/services/hi-rag-gateway/Dockerfile index b86246d6a2..335e6a563b 100644 --- a/pmoves/services/hi-rag-gateway/Dockerfile +++ b/pmoves/services/hi-rag-gateway/Dockerfile @@ -1,10 +1,11 @@ FROM python:3.11-slim WORKDIR /app -ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 +ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PYTHONPATH=/app # Base deps RUN pip install --no-cache-dir --upgrade pip -COPY hi-rag-gateway/requirements.txt hi-rag-gateway/requirements.lock ./ +# Build context is pmoves/ (contains docker-compose.yml) +COPY services/hi-rag-gateway/requirements.txt services/hi-rag-gateway/requirements.lock ./ RUN pip install --no-cache-dir -r requirements.txt # --- CUDA-enabled Torch for Qwen reranker (optional) --- @@ -15,10 +16,15 @@ ARG TORCH_SKIP_CUDA=0 RUN if [ "$TORCH_SKIP_CUDA" != "1" ]; then pip install --no-cache-dir --index-url https://download.pytorch.org/whl/${TORCH_CUDA_VERSION} torch torchvision torchaudio || true ; else echo "Skipping CUDA Torch install"; fi # Copy service code and shared libraries -COPY hi-rag-gateway/ . -COPY common/ ./services/common/ +COPY services/hi-rag-gateway/ . +COPY services/common/ ./services/common/ RUN mkdir -p /app/services && touch /app/services/__init__.py +# PMOVES Python namespace package (for ShapeStore → pmoves.chit import) +RUN mkdir -p /app/pmoves \ + && printf '"""Minimal pmoves namespace for Docker."""\n' > /app/pmoves/__init__.py +COPY chit /app/pmoves/chit + # Security: Run as non-root user RUN groupadd -r pmoves --gid=65532 && \ useradd -r -g pmoves --uid=65532 --home-dir=/app --shell=/sbin/nologin pmoves && \ diff --git a/pmoves/services/hi-rag-gateway/gateway.py b/pmoves/services/hi-rag-gateway/gateway.py index 8fc1d7ad3c..5567ca519a 100644 --- a/pmoves/services/hi-rag-gateway/gateway.py +++ b/pmoves/services/hi-rag-gateway/gateway.py @@ -1,4 +1,5 @@ import os, re, time, threading, ipaddress, math, requests, logging, json, sys, io, socket +import urllib3 from pathlib import Path from urllib.parse import urlparse @@ -531,17 +532,12 @@ def _validate_remote_image_url(raw_url: Any) -> str: return url -def _fetch_remote_image(raw_url: str, *, timeout: int = 20) -> requests.Response: - """Validate URL for SSRF and fetch with DNS-resolved IP check. +def _fetch_remote_image(raw_url: str, *, timeout: int = 20) -> urllib3.HTTPResponse: + """Validate URL for SSRF and fetch via resolved IP to prevent DNS rebinding. - Resolves DNS once and validates all IPs against private ranges before fetch. - Note: ``requests.get`` re-resolves DNS independently, so this does not fully - prevent DNS-rebinding TOCTOU attacks but raises the bar significantly. - - CodeQL alert #144 accepted risk: 5-layer defense (URL validation, scheme - check, DNS resolve, private IP block, redirect block). Only residual gap - is DNS-rebinding TOCTOU which requires attacker-controlled DNS and is - mitigated by the short TTL window. + Resolves DNS once, validates all IPs against private ranges, then connects + directly to the validated IP using urllib3 (no second DNS lookup). Sets + Host header for correct HTTP routing and server_hostname for TLS SNI. """ url = _validate_remote_image_url(raw_url) parsed = urlparse(url) @@ -567,11 +563,47 @@ def _fetch_remote_image(raw_url: str, *, timeout: int = 20) -> requests.Response ): raise HTTPException(400, f"private/internal image host blocked: {host}") - resp = requests.get(url, timeout=timeout, allow_redirects=False) - resp.raise_for_status() - if 300 <= resp.status_code < 400: - raise HTTPException(400, f"redirect responses are not allowed for image URL: {url}") - return resp + req_path = (parsed.path or "/") + (f"?{parsed.query}" if parsed.query else "") + pool_timeout = urllib3.util.Timeout(connect=10, read=timeout) + + # Deduplicate resolved IPs while preserving order + seen_ips: set = set() + unique_ips: list = [] + for _, _, _, _, sockaddr in addrs: + ip = sockaddr[0] + if ip not in seen_ips: + seen_ips.add(ip) + unique_ips.append(ip) + + # Try each resolved address (handles dual-stack AAAA/A and round-robin DNS) + last_exc: Optional[Exception] = None + for resolved_ip in unique_ips: + if parsed.scheme == "https": + pool = urllib3.HTTPSConnectionPool( + resolved_ip, port=port, + timeout=pool_timeout, + server_hostname=host, + ) + else: + pool = urllib3.HTTPConnectionPool( + resolved_ip, port=port, + timeout=pool_timeout, + ) + try: + http_resp = pool.request("GET", req_path, headers={"Host": host}, redirect=False) + except Exception as exc: + pool.close() + last_exc = exc + continue + pool.close() + if http_resp.status >= 400: + raise HTTPException(400, f"remote image fetch failed with HTTP {http_resp.status}") + if 300 <= http_resp.status < 400: + raise HTTPException(400, f"redirect responses are not allowed for image URL: {url}") + return http_resp + + # All resolved addresses failed + raise HTTPException(400, f"all resolved addresses unreachable for {host}: {last_exc}") def run_query(query, namespace, k=8, alpha=0.7, graph_boost=GRAPH_BOOST, entity_types=None): @@ -863,10 +895,13 @@ def geometry_decode_image(body: Dict[str, Any], _=Depends(require_tailscale)): for url in images: try: r = _fetch_remote_image(url) - except requests.RequestException as e: + except (urllib3.exceptions.HTTPError, OSError) as e: raise HTTPException(502, f"failed to fetch image: {e}") + image_bytes = r.data + if not image_bytes: + raise HTTPException(400, f"remote image returned empty body for {url}") try: - img = Image.open(io.BytesIO(r.content)).convert("RGB") + img = Image.open(io.BytesIO(image_bytes)).convert("RGB") except Exception as e: raise HTTPException(400, f"invalid image payload for {url}: {e}") img_list.append(img) diff --git a/pmoves/services/pmoves-yt/yt.py b/pmoves/services/pmoves-yt/yt.py index b21599683f..a87e1c7cd7 100644 --- a/pmoves/services/pmoves-yt/yt.py +++ b/pmoves/services/pmoves-yt/yt.py @@ -1274,7 +1274,7 @@ def _download_with_yt_dlp( s3_url = upload_to_s3(outpath, bucket, raw_key) thumb = None for ext in ('.jpg', '.png', '.webp'): - cand = os.path.join(str(vid_dir), f"{vid}{ext}") + cand = os.path.join(str(vid_dir), f"{vid}{ext}") # CodeQL path-injection: vid from yt-dlp info['id'] — constrained alphanumeric if os.path.exists(cand): thumb_key = f"{base}/thumb{ext}" thumb = upload_to_s3(cand, bucket, thumb_key) @@ -1430,13 +1430,13 @@ def _download_with_companion( if "webm" in mime: ext = "webm" base = base_prefix(video_id, platform) - vid_dir = YT_TEMP_ROOT / video_id + vid_dir = YT_TEMP_ROOT / video_id # CodeQL path-injection: sanitized by _safe_video_id (basename + regex allowlist) vid_dir.mkdir(parents=True, exist_ok=True) - tmp_path = vid_dir / f"{video_id}.{ext}" + tmp_path = vid_dir / f"{video_id}.{ext}" # CodeQL path-injection: sanitized by _safe_video_id (basename + regex allowlist) try: with requests.get(download_url, stream=True, timeout=120) as r: r.raise_for_status() - with open(tmp_path, "wb") as fh: + with open(tmp_path, "wb") as fh: # CodeQL path-injection: sanitized by _safe_video_id (basename + regex allowlist) for chunk in r.iter_content(1 << 20): if chunk: fh.write(chunk) @@ -1457,7 +1457,7 @@ def _download_with_companion( r_thumb = requests.get(thumb_url, timeout=20) r_thumb.raise_for_status() thumb_path = vid_dir / f"{video_id}_thumb.jpg" - with open(thumb_path, "wb") as tfh: + with open(thumb_path, "wb") as tfh: # CodeQL path-injection: sanitized by _safe_video_id (basename + regex allowlist) tfh.write(r_thumb.content) thumb = upload_to_s3(str(thumb_path), bucket, f"{base}/thumb.jpg") break @@ -1572,13 +1572,13 @@ def _download_with_invidious( if 'webm' in content_type: ext = 'webm' base = base_prefix(video_id, platform_key) - vid_dir = YT_TEMP_ROOT / video_id + vid_dir = YT_TEMP_ROOT / video_id # CodeQL path-injection: sanitized by _safe_video_id (basename + regex allowlist) vid_dir.mkdir(parents=True, exist_ok=True) - tmp_path = vid_dir / f"{video_id}.{ext}" + tmp_path = vid_dir / f"{video_id}.{ext}" # CodeQL path-injection: sanitized by _safe_video_id (basename + regex allowlist) try: with requests.get(download_url, stream=True, timeout=120) as r: r.raise_for_status() - with open(tmp_path, 'wb') as fh: + with open(tmp_path, 'wb') as fh: # CodeQL path-injection: sanitized by _safe_video_id (basename + regex allowlist) for chunk in r.iter_content(1 << 20): if chunk: fh.write(chunk) @@ -1597,7 +1597,7 @@ def _download_with_invidious( resp.raise_for_status() thumb_ext = 'jpg' thumb_path = vid_dir / f"{video_id}_thumb.{thumb_ext}" - with open(thumb_path, 'wb') as tfh: + with open(thumb_path, 'wb') as tfh: # CodeQL path-injection: sanitized by _safe_video_id (basename + regex allowlist) tfh.write(resp.content) thumb_key = f"{base}/thumb.{thumb_ext}" thumb_s3 = upload_to_s3(str(thumb_path), bucket, thumb_key) @@ -1773,11 +1773,11 @@ def yt_download(body: Dict[str,Any] = Body(...)): archive_enabled = bool(yt_options.get('use_download_archive', YT_ENABLE_DOWNLOAD_ARCHIVE)) archive_path_value = yt_options.get('download_archive', YT_DOWNLOAD_ARCHIVE) if archive_enabled and archive_path_value: - safe_name = os.path.basename(archive_path_value) + safe_name = os.path.basename(archive_path_value) # CodeQL path-injection: sanitized by os.path.basename — only filename component retained if not safe_name: safe_name = "download-archive.txt" archive_path = YT_ARCHIVE_DIR / safe_name - archive_path.parent.mkdir(parents=True, exist_ok=True) + archive_path.parent.mkdir(parents=True, exist_ok=True) # CodeQL path-injection: sanitized by os.path.basename above ydl_opts['download_archive'] = str(archive_path) subtitle_langs = yt_options.get('subtitle_langs', None) diff --git a/pmoves/tools/chit_credential_demo.py b/pmoves/tools/chit_credential_demo.py index c6f0977e31..a3fac86b8b 100644 --- a/pmoves/tools/chit_credential_demo.py +++ b/pmoves/tools/chit_credential_demo.py @@ -118,9 +118,8 @@ def cmd_verify(args: argparse.Namespace) -> int: print(f"\nDecoded {len(secrets)} key(s).") for key in sorted(secrets): val = secrets[key] - # Mask secret values — show only first 4 chars - display = val[:4] + "****" if len(val) > 4 else "****" - print(f" {key} = {display}") + # Mask secret values — show only key names, never partial values + print(f" {key} = ****") return 0 @@ -235,7 +234,7 @@ def cmd_report(args: argparse.Namespace) -> int: if findings: print(f"\nPLAINTEXT CREDENTIALS FOUND: {len(findings)}") for rel_path, line_no, desc, snippet in findings: - print(f" {rel_path}:{line_no} [{desc}] {snippet}") + print(f" {rel_path}:{line_no} [{desc}] {snippet}") # CodeQL clear-text-logging: intentional — diagnostic tool for credential audit return 1 else: print(f"\nNo plaintext credentials found.") diff --git a/pmoves/ui/lib/serviceHealth.ts b/pmoves/ui/lib/serviceHealth.ts index 562bec1655..20a980bb39 100644 --- a/pmoves/ui/lib/serviceHealth.ts +++ b/pmoves/ui/lib/serviceHealth.ts @@ -51,10 +51,11 @@ export async function probeService( }; } - try { - const controller = new AbortController(); - const timeoutId = setTimeout(() => controller.abort(), timeout); + const safeTimeout = Math.min(Math.max(timeout, 1000), 60_000); + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), safeTimeout); + try { const response = await fetch(service.healthCheck, { method: 'GET', signal: controller.signal, @@ -62,8 +63,6 @@ export async function probeService( cache: 'no-store', }); - clearTimeout(timeoutId); - const responseTime = performance.now() - startTime; return { @@ -83,6 +82,8 @@ export async function probeService( lastCheck: new Date(), error: error instanceof Error ? error.message : 'Unknown error', }; + } finally { + clearTimeout(timeoutId); } }