WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-08-17 18:25:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] publish: split 379-line monolith into scan/build/cloud modules + memoize scan by source hash

Browse Source
This commit is contained in:
ciregenz
2026-06-20 03:06:44 -07:00
parent b6e230e7ad
commit ee5b5c2cea
7 changed files with 495 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/outputs/outputs.py
+4 -4
View File
@@ -16,10 +16,10 @@ from backend.apps.outputs.models import (
PublishResult, PublishReview,
)
from backend.apps.outputs.executor import execute_backend_code, get_code_warnings
from backend.apps.outputs.publish import (
scan_for_publish, quick_ast_gate, build_static, collect_bundle,
upload_to_cloud, unpublish_from_cloud, slugify, PublishError,
)
from backend.apps.outputs.publish_common import slugify, PublishError
from backend.apps.outputs.publish_scan import scan_for_publish, quick_ast_gate
from backend.apps.outputs.publish_build import build_static, collect_bundle
from backend.apps.outputs.publish_cloud import upload_to_cloud, unpublish_from_cloud
from backend.apps.outputs.view_builder_templates import (
VIEW_TEMPLATE_FILES,
load_app_builder_skill,
backend/apps/outputs/publish.py
-379
View File
@@ -1,379 +0,0 @@
"""App publishing: build the static bundle, scan it, ship it to the cloud host.
A finished app is static: webapp-mode builds to `frontend/dist` via the bundled
node (same node the runtime spawns vite with); flat-mode is already `index.html`.
The optional `backend.py` is the sandboxed data-shaping kind, so the cloud edge
can run it on a shared sandbox. Scanning runs here on the user's OWN creds, so it
costs us nothing and the code never leaves the machine until they choose to ship.
Layering note: this lives under `outputs/` (below `swarm/`), so it reuses the
local `get_code_warnings` and defines its own review model rather than importing
`swarm.review` upward. The JSON shape matches the frontend `ReviewSummary`."""
from __future__ import annotations
import asyncio
import io
import json
import logging
import os
import re
import shutil
import tarfile
from typing import Literal, Optional
import httpx
from backend.apps.outputs.executor import get_code_warnings
from backend.apps.outputs.models import Output, PublishReview
from backend.apps.outputs.workspace_io import _WALK_SKIP_DIRS
from backend.apps.settings.credentials import OPENSWARM_DEFAULT_PROXY_URL
from backend.config.paths import OUTPUTS_WORKSPACE_DIR
logger = logging.getLogger(__name__)
_BUILD_TIMEOUT = 180 # vite build on a cold-ish node_modules can be slow
_SCAN_CODE_BUDGET = 60_000 # chars of source we hand the aux model
_MAX_BUNDLE_FILE = 25 * 1024 * 1024
_SCAN_EXTS = (".py", ".html", ".ts", ".tsx", ".js", ".jsx", ".vue", ".svelte", ".css")
_SCAN_SYSTEM_PROMPT = (
"You are a security reviewer for a no-code app host. The app below will be "
"served publicly at a *.openswarm.host subdomain. Read the source and report "
"only concrete, real risks a reviewer would act on: hardcoded secrets or API "
"keys, phishing or credential-harvesting forms, sending user data to a "
"third-party endpoint, obvious XSS or injection, or anything malicious. Do "
"NOT nitpick style or speculate. Reply ONLY with JSON: "
'{"severity": "clean|warn|block", "findings": ["short dev-readable line", ...]}. '
"Use block only for clearly malicious or credential-harvesting code. Empty "
"findings means clean."
)
class PublishError(Exception):
"""User-facing publish failure; message is safe to show in a toast."""
def slugify(name: str) -> str:
"""A url-safe slug hint from the app name; the cloud guarantees uniqueness."""
s = re.sub(r"[^a-z0-9]+", "-", (name or "app").lower()).strip("-")
s = s[:32].strip("-")
return s or "app"
def is_webapp(output: Output) -> bool:
return bool(output.workspace_id)
def _workspace_dir(output: Output) -> str:
return os.path.join(OUTPUTS_WORKSPACE_DIR, output.workspace_id or "")
def _node_bin() -> Optional[str]:
return os.environ.get("OPENSWARM_NODE_PATH") or shutil.which("node")
# --- source collection (for scanning) ---------------------------------------
def _collect_source(output: Output) -> dict[str, str]:
"""Gather human-readable source text for the scan. Flat apps come from the
files dict; webapp apps walk the workspace skipping node_modules/.venv/dist."""
src: dict[str, str] = {}
for name, content in (output.files or {}).items():
if name.lower().endswith(_SCAN_EXTS):
src[name] = content
if is_webapp(output):
root = _workspace_dir(output)
for base, _dirs, fnames in os.walk(root):
_dirs[:] = [d for d in _dirs if d not in _WALK_SKIP_DIRS]
for fn in fnames:
if not fn.lower().endswith(_SCAN_EXTS):
continue
full = os.path.join(base, fn)
if os.path.islink(full):
continue
try:
if os.path.getsize(full) > 512 * 1024:
continue
with open(full, "r", encoding="utf-8", errors="replace") as f:
rel = os.path.relpath(full, root).replace(os.sep, "/")
src[rel] = f.read()
except OSError:
continue
return src
def _scan_blob(src: dict[str, str]) -> str:
parts: list[str] = []
total = 0
for path, code in src.items():
chunk = f"=== {path} ===\n{code}\n"
if total + len(chunk) > _SCAN_CODE_BUDGET:
chunk = chunk[: max(0, _SCAN_CODE_BUDGET - total)]
parts.append(chunk)
total += len(chunk)
if total >= _SCAN_CODE_BUDGET:
break
return "".join(parts)
def _ast_findings(src: dict[str, str]) -> tuple[list[str], list[str]]:
findings: list[str] = []
scanned: list[str] = []
for path, code in src.items():
if path.lower().endswith(".py"):
scanned.append(path)
for w in get_code_warnings(code):
findings.append(f"{path}: {w}")
return findings, scanned
async def _llm_findings(src: dict[str, str], settings) -> tuple[list[str], str]:
"""Aux-tier semantic pass. Best-effort: if no aux model is configured or the
call fails, return clean so the AST pass still gates. Runs on the user's creds."""
blob = _scan_blob(src)
if not blob.strip():
return [], "clean"
from backend.apps.agents.providers.registry import resolve_aux_model
from backend.apps.settings.credentials import get_anthropic_client_for_model
from backend.apps.agents.core.aux_llm import _safe_resp_text
try:
model, _base = await resolve_aux_model(settings, preferred_tier="haiku")
except Exception:
return [], "clean"
client = get_anthropic_client_for_model(settings, model)
try:
resp = await client.messages.create(
model=model,
max_tokens=1200,
system=_SCAN_SYSTEM_PROMPT,
messages=[{"role": "user", "content": blob}],
)
except Exception:
logger.exception("publish LLM scan call failed; AST-only result stands")
return [], "clean"
text = _safe_resp_text(resp).strip()
if text.startswith("```"):
text = text.split("\n", 1)[1] if "\n" in text else text[3:]
if text.endswith("```"):
text = text[:-3]
try:
parsed = json.loads(text)
except (json.JSONDecodeError, ValueError):
return [], "clean"
findings = [str(f) for f in parsed.get("findings", []) if str(f).strip()][:20]
severity = parsed.get("severity", "clean")
if severity not in ("clean", "warn", "block"):
severity = "warn" if findings else "clean"
return findings, severity
async def scan_for_publish(output: Output, settings) -> PublishReview:
src = _collect_source(output)
ast_findings, scanned = _ast_findings(src)
llm_findings, llm_sev = await _llm_findings(src, settings)
findings = ast_findings + llm_findings
verdict: Literal["clean", "warn", "block"] = "clean"
if findings:
verdict = "warn"
if llm_sev == "block":
verdict = "block"
return PublishReview(
verdict=verdict,
findings=findings,
scanned_files=scanned or sorted(src.keys()),
)
def quick_ast_gate(output: Output) -> list[str]:
"""Cheap, free safety net used by /publish when force is not set: flags the
AST-visible 'runs code outside the sandbox' findings without an LLM call."""
findings, _ = _ast_findings(_collect_source(output))
return findings
# --- build + bundle ----------------------------------------------------------
def _safe_build_config(fe: str) -> tuple[list[str], Optional[str]]:
"""vite-plugin-terminal injects a dev-only `virtual:terminal` module that
breaks `vite build` in older workspaces (the template later gated it to dev,
but apps seeded before that still carry the ungated plugin). Build against a
temp config that makes that plugin a no-op (Vite drops null plugins) so ANY
workspace builds clean. The user's own vite.config is never touched.
Returns (extra build args, temp-config path to delete) or ([], None)."""
cfg_name = next(
(n for n in ("vite.config.ts", "vite.config.js", "vite.config.mjs")
if os.path.exists(os.path.join(fe, n))),
None,
)
if not cfg_name:
return [], None
with open(os.path.join(fe, cfg_name), "r", encoding="utf-8") as f:
content = f.read()
if "vite-plugin-terminal" not in content:
return [], None
patched = re.sub(
r"import\s+terminal\s+from\s+['\"]vite-plugin-terminal['\"];?",
"const terminal = () => null;",
content,
)
ext = os.path.splitext(cfg_name)[1]
temp_name = f"vite.config.openswarm-publish{ext}"
with open(os.path.join(fe, temp_name), "w", encoding="utf-8") as f:
f.write(patched)
return ["--config", temp_name], os.path.join(fe, temp_name)
async def build_static(output: Output) -> Optional[str]:
"""Webapp apps -> build `frontend/dist`, return its path. Flat apps need no
build (the files dict is the artifact), return None. Raises PublishError with
a user-safe message on any failure."""
if not is_webapp(output):
return None
fe = os.path.join(_workspace_dir(output), "frontend")
vite = os.path.join(fe, "node_modules", "vite", "bin", "vite.js")
node = _node_bin()
if not node or not os.path.exists(vite):
raise PublishError(
"This app isn't set up to build yet. Open it once in the editor, then try publishing again."
)
config_args, temp_cfg = _safe_build_config(fe)
proc = await asyncio.create_subprocess_exec(
node, "node_modules/vite/bin/vite.js", "build", *config_args,
cwd=fe,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env={**os.environ, "NODE_ENV": "production"},
)
try:
_out, err = await asyncio.wait_for(proc.communicate(), timeout=_BUILD_TIMEOUT)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise PublishError("Building your app took too long and was stopped.")
finally:
if temp_cfg:
try:
os.remove(temp_cfg)
except OSError:
pass
if proc.returncode != 0:
logger.error("vite build failed (%s): %s", output.id, err.decode(errors="replace")[-2000:])
raise PublishError("We couldn't build your app. Make sure it runs in the editor, then try again.")
dist = os.path.join(fe, "dist")
if not os.path.isdir(dist):
raise PublishError("The build finished but produced no files.")
return dist
_SECRET_KEY_EXTS = (".pem", ".key", ".p12", ".pfx", ".keystore")
def _is_secret_file(rel_path: str) -> bool:
"""This bundle is served publicly, so anything secret-shaped must never make it
in. dotenv files and private-key material are the realistic leaks; the webapp
path already ships only the built dist, this also covers a hand-built flat app."""
base = rel_path.rsplit("/", 1)[-1].lower()
return (
base == ".env"
or base.startswith(".env.")
or base.endswith(_SECRET_KEY_EXTS)
or base in (".npmrc", ".git-credentials", ".htpasswd")
)
def collect_bundle(output: Output, dist_dir: Optional[str]) -> bytes:
"""tar.gz of what the cloud should host. Webapp -> the built dist tree.
Flat -> the files dict, including backend.py (the edge runs it on the shared
sandbox; the edge refuses to serve .py as a static file). Secret-shaped files
(.env, private keys) are dropped: a published bundle is world-readable."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
if dist_dir:
for root, _dirs, files in os.walk(dist_dir):
for fn in files:
full = os.path.join(root, fn)
if os.path.islink(full):
continue
rel = os.path.relpath(full, dist_dir).replace(os.sep, "/")
if _is_secret_file(rel):
continue
try:
if os.path.getsize(full) > _MAX_BUNDLE_FILE:
continue
except OSError:
continue
tar.add(full, arcname=rel)
else:
for name, content in (output.files or {}).items():
rel = name.replace(os.sep, "/")
if _is_secret_file(rel):
continue
data = content.encode("utf-8")
if len(data) > _MAX_BUNDLE_FILE:
continue
info = tarfile.TarInfo(name=rel)
info.size = len(data)
tar.addfile(info, io.BytesIO(data))
return buf.getvalue()
# --- cloud client ------------------------------------------------------------
def _cloud_auth(settings) -> tuple[Optional[str], str]:
"""Publish works for ANY signed-in account, so read the bearer directly
rather than proxy_auth (which only yields a token in pro/free-trial modes).
Matches the cloud's requireAuthedUser gate."""
base = (getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
token = getattr(settings, "openswarm_bearer_token", None)
return token, base
def _safe_detail(resp: httpx.Response, fallback: str) -> str:
try:
body = resp.json()
msg = body.get("message") or body.get("error")
if isinstance(msg, str) and msg:
return msg
except Exception:
pass
return fallback
async def upload_to_cloud(
settings, *, output_id: str, name: str, slug_hint: str, bundle: bytes, override: bool
) -> dict:
token, base = _cloud_auth(settings)
if not token:
raise PublishError("Sign in to your OpenSwarm account to publish apps.")
try:
async with httpx.AsyncClient(timeout=120.0) as client:
r = await client.post(
f"{base}/api/apps/publish",
headers={"Authorization": f"Bearer {token}"},
# output_id lets the cloud reuse this app's slug on republish instead
# of minting a duplicate; override marks a publish past a non-clean scan.
data={"name": name, "slug": slug_hint, "output_id": output_id, "override": "1" if override else "0"},
files={"bundle": ("app.tar.gz", bundle, "application/gzip")},
)
except httpx.HTTPError:
raise PublishError("Couldn't reach the publishing service. Check your connection and try again.")
if r.status_code >= 400:
raise PublishError(_safe_detail(r, "Publishing failed. Please try again."))
return r.json()
async def unpublish_from_cloud(settings, slug: str) -> None:
token, base = _cloud_auth(settings)
if not token:
raise PublishError("Sign in to your OpenSwarm account to manage published apps.")
try:
async with httpx.AsyncClient(timeout=30.0) as client:
r = await client.post(
f"{base}/api/apps/{slug}/delete",
headers={"Authorization": f"Bearer {token}"},
)
except httpx.HTTPError:
raise PublishError("Couldn't reach the publishing service. Check your connection and try again.")
if r.status_code >= 400 and r.status_code != 404:
raise PublishError(_safe_detail(r, "Couldn't unpublish. Please try again."))
backend/apps/outputs/publish_build.py
+148
View File
@@ -0,0 +1,148 @@
"""Build + bundle the static artifact the cloud will host. Webapp-mode runs the
bundled node on `vite build`; flat-mode is already the artifact. Secret-shaped
files (.env, private keys) are dropped because a published bundle is world-readable."""
from __future__ import annotations
import asyncio
import io
import logging
import os
import re
import shutil
import tarfile
from typing import Optional
from backend.apps.outputs.models import Output
from backend.apps.outputs.publish_common import PublishError, is_webapp, workspace_dir
logger = logging.getLogger(__name__)
_BUILD_TIMEOUT = 180 # vite build on a cold-ish node_modules can be slow
_MAX_BUNDLE_FILE = 25 * 1024 * 1024
_SECRET_KEY_EXTS = (".pem", ".key", ".p12", ".pfx", ".keystore")
def _node_bin() -> Optional[str]:
return os.environ.get("OPENSWARM_NODE_PATH") or shutil.which("node")
def _safe_build_config(fe: str) -> tuple[list[str], Optional[str]]:
"""vite-plugin-terminal injects a dev-only `virtual:terminal` module that
breaks `vite build` in older workspaces (the template later gated it to dev,
but apps seeded before that still carry the ungated plugin). Build against a
temp config that makes that plugin a no-op (Vite drops null plugins) so ANY
workspace builds clean. The user's own vite.config is never touched.
Returns (extra build args, temp-config path to delete) or ([], None)."""
cfg_name = next(
(n for n in ("vite.config.ts", "vite.config.js", "vite.config.mjs")
if os.path.exists(os.path.join(fe, n))),
None,
)
if not cfg_name:
return [], None
with open(os.path.join(fe, cfg_name), "r", encoding="utf-8") as f:
content = f.read()
if "vite-plugin-terminal" not in content:
return [], None
patched = re.sub(
r"import\s+terminal\s+from\s+['\"]vite-plugin-terminal['\"];?",
"const terminal = () => null;",
content,
)
ext = os.path.splitext(cfg_name)[1]
temp_name = f"vite.config.openswarm-publish{ext}"
with open(os.path.join(fe, temp_name), "w", encoding="utf-8") as f:
f.write(patched)
return ["--config", temp_name], os.path.join(fe, temp_name)
async def build_static(output: Output) -> Optional[str]:
"""Webapp apps -> build `frontend/dist`, return its path. Flat apps need no
build (the files dict is the artifact), return None. Raises PublishError with
a user-safe message on any failure."""
if not is_webapp(output):
return None
fe = os.path.join(workspace_dir(output), "frontend")
vite = os.path.join(fe, "node_modules", "vite", "bin", "vite.js")
node = _node_bin()
if not node or not os.path.exists(vite):
raise PublishError(
"This app isn't set up to build yet. Open it once in the editor, then try publishing again."
)
config_args, temp_cfg = _safe_build_config(fe)
proc = await asyncio.create_subprocess_exec(
node, "node_modules/vite/bin/vite.js", "build", *config_args,
cwd=fe,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env={**os.environ, "NODE_ENV": "production"},
)
try:
_out, err = await asyncio.wait_for(proc.communicate(), timeout=_BUILD_TIMEOUT)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise PublishError("Building your app took too long and was stopped.")
finally:
if temp_cfg:
try:
os.remove(temp_cfg)
except OSError:
pass
if proc.returncode != 0:
logger.error("vite build failed (%s): %s", output.id, err.decode(errors="replace")[-2000:])
raise PublishError("We couldn't build your app. Make sure it runs in the editor, then try again.")
dist = os.path.join(fe, "dist")
if not os.path.isdir(dist):
raise PublishError("The build finished but produced no files.")
return dist
def _is_secret_file(rel_path: str) -> bool:
"""This bundle is served publicly, so anything secret-shaped must never make it
in. dotenv files and private-key material are the realistic leaks; the webapp
path already ships only the built dist, this also covers a hand-built flat app."""
base = rel_path.rsplit("/", 1)[-1].lower()
return (
base == ".env"
or base.startswith(".env.")
or base.endswith(_SECRET_KEY_EXTS)
or base in (".npmrc", ".git-credentials", ".htpasswd")
)
def collect_bundle(output: Output, dist_dir: Optional[str]) -> bytes:
"""tar.gz of what the cloud should host. Webapp -> the built dist tree.
Flat -> the files dict, including backend.py (the edge runs it on the shared
sandbox; the edge refuses to serve .py as a static file). Secret-shaped files
(.env, private keys) are dropped: a published bundle is world-readable."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
if dist_dir:
for root, _dirs, files in os.walk(dist_dir):
for fn in files:
full = os.path.join(root, fn)
if os.path.islink(full):
continue
rel = os.path.relpath(full, dist_dir).replace(os.sep, "/")
if _is_secret_file(rel):
continue
try:
if os.path.getsize(full) > _MAX_BUNDLE_FILE:
continue
except OSError:
continue
tar.add(full, arcname=rel)
else:
for name, content in (output.files or {}).items():
rel = name.replace(os.sep, "/")
if _is_secret_file(rel):
continue
data = content.encode("utf-8")
if len(data) > _MAX_BUNDLE_FILE:
continue
info = tarfile.TarInfo(name=rel)
info.size = len(data)
tar.addfile(info, io.BytesIO(data))
return buf.getvalue()
backend/apps/outputs/publish_cloud.py
+67
View File
@@ -0,0 +1,67 @@
"""Cloud client for the publish pipeline: ships the bundle to the host and takes
it back down. Reads the bearer directly (publish works for any signed-in account,
not just pro/free-trial), matching the cloud's requireAuthedUser gate."""
from __future__ import annotations
from typing import Optional
import httpx
from backend.apps.outputs.publish_common import PublishError
from backend.apps.settings.credentials import OPENSWARM_DEFAULT_PROXY_URL
def _cloud_auth(settings) -> tuple[Optional[str], str]:
base = (getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
token = getattr(settings, "openswarm_bearer_token", None)
return token, base
def _safe_detail(resp: httpx.Response, fallback: str) -> str:
try:
body = resp.json()
msg = body.get("message") or body.get("error")
if isinstance(msg, str) and msg:
return msg
except Exception:
pass
return fallback
async def upload_to_cloud(
settings, *, output_id: str, name: str, slug_hint: str, bundle: bytes, override: bool
) -> dict:
token, base = _cloud_auth(settings)
if not token:
raise PublishError("Sign in to your OpenSwarm account to publish apps.")
try:
async with httpx.AsyncClient(timeout=120.0) as client:
r = await client.post(
f"{base}/api/apps/publish",
headers={"Authorization": f"Bearer {token}"},
# output_id lets the cloud reuse this app's slug on republish instead
# of minting a duplicate; override marks a publish past a non-clean scan.
data={"name": name, "slug": slug_hint, "output_id": output_id, "override": "1" if override else "0"},
files={"bundle": ("app.tar.gz", bundle, "application/gzip")},
)
except httpx.HTTPError:
raise PublishError("Couldn't reach the publishing service. Check your connection and try again.")
if r.status_code >= 400:
raise PublishError(_safe_detail(r, "Publishing failed. Please try again."))
return r.json()
async def unpublish_from_cloud(settings, slug: str) -> None:
token, base = _cloud_auth(settings)
if not token:
raise PublishError("Sign in to your OpenSwarm account to manage published apps.")
try:
async with httpx.AsyncClient(timeout=30.0) as client:
r = await client.post(
f"{base}/api/apps/{slug}/delete",
headers={"Authorization": f"Bearer {token}"},
)
except httpx.HTTPError:
raise PublishError("Couldn't reach the publishing service. Check your connection and try again.")
if r.status_code >= 400 and r.status_code != 404:
raise PublishError(_safe_detail(r, "Couldn't unpublish. Please try again."))
backend/apps/outputs/publish_common.py
+29
View File
@@ -0,0 +1,29 @@
"""Shared low-level bits for the publish pipeline (scan / build / cloud). Kept
tiny and dependency-light so scan, build, and cloud_client can all lean on it
without reaching sideways into each other."""
from __future__ import annotations
import os
import re
from backend.apps.outputs.models import Output
from backend.config.paths import OUTPUTS_WORKSPACE_DIR
class PublishError(Exception):
"""User-facing publish failure; message is safe to show in a toast."""
def slugify(name: str) -> str:
"""A url-safe slug hint from the app name; the cloud guarantees uniqueness."""
s = re.sub(r"[^a-z0-9]+", "-", (name or "app").lower()).strip("-")
s = s[:32].strip("-")
return s or "app"
def is_webapp(output: Output) -> bool:
return bool(output.workspace_id)
def workspace_dir(output: Output) -> str:
return os.path.join(OUTPUTS_WORKSPACE_DIR, output.workspace_id or "")
backend/apps/outputs/publish_scan.py
+179
View File
@@ -0,0 +1,179 @@
"""Pre-publish security scan: a free AST pass (reuses the executor's
`get_code_warnings`) plus a best-effort aux-LLM semantic pass, both on the user's
OWN creds so it costs us nothing and the code never leaves the machine until they
ship. The JSON shape matches the frontend `ReviewSummary`.
The full scan (`scan_for_publish`) is memoized on a hash of the collected source:
reopening the publish modal on unchanged code returns the cached review instead of
billing the user's aux model again."""
from __future__ import annotations
import hashlib
import json
import logging
import os
from collections import OrderedDict
from typing import Literal
from backend.apps.outputs.executor import get_code_warnings
from backend.apps.outputs.models import Output, PublishReview
from backend.apps.outputs.publish_common import is_webapp, workspace_dir
from backend.apps.outputs.workspace_io import _WALK_SKIP_DIRS
logger = logging.getLogger(__name__)
_SCAN_CODE_BUDGET = 60_000 # chars of source we hand the aux model
_SCAN_EXTS = (".py", ".html", ".ts", ".tsx", ".js", ".jsx", ".vue", ".svelte", ".css")
_MEMO_MAX = 32
_SCAN_SYSTEM_PROMPT = (
"You are a security reviewer for a no-code app host. The app below will be "
"served publicly at a *.openswarm.host subdomain. Read the source and report "
"only concrete, real risks a reviewer would act on: hardcoded secrets or API "
"keys, phishing or credential-harvesting forms, sending user data to a "
"third-party endpoint, obvious XSS or injection, or anything malicious. Do "
"NOT nitpick style or speculate. Reply ONLY with JSON: "
'{"severity": "clean|warn|block", "findings": ["short dev-readable line", ...]}. '
"Use block only for clearly malicious or credential-harvesting code. Empty "
"findings means clean."
)
# slug-content-hash -> PublishReview, so a reopened modal doesn't re-bill the LLM.
_memo: "OrderedDict[str, PublishReview]" = OrderedDict()
def _collect_source(output: Output) -> dict[str, str]:
"""Gather human-readable source text for the scan. Flat apps come from the
files dict; webapp apps walk the workspace skipping node_modules/.venv/dist."""
src: dict[str, str] = {}
for name, content in (output.files or {}).items():
if name.lower().endswith(_SCAN_EXTS):
src[name] = content
if is_webapp(output):
root = workspace_dir(output)
for base, _dirs, fnames in os.walk(root):
_dirs[:] = [d for d in _dirs if d not in _WALK_SKIP_DIRS]
for fn in fnames:
if not fn.lower().endswith(_SCAN_EXTS):
continue
full = os.path.join(base, fn)
if os.path.islink(full):
continue
try:
if os.path.getsize(full) > 512 * 1024:
continue
with open(full, "r", encoding="utf-8", errors="replace") as f:
rel = os.path.relpath(full, root).replace(os.sep, "/")
src[rel] = f.read()
except OSError:
continue
return src
def _source_hash(src: dict[str, str]) -> str:
h = hashlib.sha256()
for path in sorted(src):
h.update(path.encode("utf-8"))
h.update(b"\0")
h.update(src[path].encode("utf-8", errors="replace"))
h.update(b"\0")
return h.hexdigest()
def _scan_blob(src: dict[str, str]) -> str:
parts: list[str] = []
total = 0
for path, code in src.items():
chunk = f"=== {path} ===\n{code}\n"
if total + len(chunk) > _SCAN_CODE_BUDGET:
chunk = chunk[: max(0, _SCAN_CODE_BUDGET - total)]
parts.append(chunk)
total += len(chunk)
if total >= _SCAN_CODE_BUDGET:
break
return "".join(parts)
def _ast_findings(src: dict[str, str]) -> tuple[list[str], list[str]]:
findings: list[str] = []
scanned: list[str] = []
for path, code in src.items():
if path.lower().endswith(".py"):
scanned.append(path)
for w in get_code_warnings(code):
findings.append(f"{path}: {w}")
return findings, scanned
async def _llm_findings(src: dict[str, str], settings) -> tuple[list[str], str]:
"""Aux-tier semantic pass. Best-effort: if no aux model is configured or the
call fails, return clean so the AST pass still gates. Runs on the user's creds."""
blob = _scan_blob(src)
if not blob.strip():
return [], "clean"
from backend.apps.agents.providers.registry import resolve_aux_model
from backend.apps.settings.credentials import get_anthropic_client_for_model
from backend.apps.agents.core.aux_llm import _safe_resp_text
try:
model, _base = await resolve_aux_model(settings, preferred_tier="haiku")
except Exception:
return [], "clean"
client = get_anthropic_client_for_model(settings, model)
try:
resp = await client.messages.create(
model=model,
max_tokens=1200,
system=_SCAN_SYSTEM_PROMPT,
messages=[{"role": "user", "content": blob}],
)
except Exception:
logger.exception("publish LLM scan call failed; AST-only result stands")
return [], "clean"
text = _safe_resp_text(resp).strip()
if text.startswith("```"):
text = text.split("\n", 1)[1] if "\n" in text else text[3:]
if text.endswith("```"):
text = text[:-3]
try:
parsed = json.loads(text)
except (json.JSONDecodeError, ValueError):
return [], "clean"
findings = [str(f) for f in parsed.get("findings", []) if str(f).strip()][:20]
severity = parsed.get("severity", "clean")
if severity not in ("clean", "warn", "block"):
severity = "warn" if findings else "clean"
return findings, severity
async def scan_for_publish(output: Output, settings) -> PublishReview:
src = _collect_source(output)
key = _source_hash(src)
cached = _memo.get(key)
if cached is not None:
_memo.move_to_end(key)
return cached
ast_findings, scanned = _ast_findings(src)
llm_findings, llm_sev = await _llm_findings(src, settings)
findings = ast_findings + llm_findings
verdict: Literal["clean", "warn", "block"] = "clean"
if findings:
verdict = "warn"
if llm_sev == "block":
verdict = "block"
review = PublishReview(
verdict=verdict,
findings=findings,
scanned_files=scanned or sorted(src.keys()),
)
_memo[key] = review
_memo.move_to_end(key)
while len(_memo) > _MEMO_MAX:
_memo.popitem(last=False)
return review
def quick_ast_gate(output: Output) -> list[str]:
"""Cheap, free safety net used by /publish when force is not set: flags the
AST-visible 'runs code outside the sandbox' findings without an LLM call."""
findings, _ = _ast_findings(_collect_source(output))
return findings
backend/tests/test_publish.py
+68 -19
View File
@@ -8,9 +8,10 @@ What this proves:
and stays silent for allowlist-only code.
3. _collect_source picks up flat files and skips binary/non-source.
4. collect_bundle (flat) tars exactly the files dict; (webapp) tars a dist tree
and skips symlinks.
and skips symlinks; secret-shaped files never make it into a public bundle.
5. scan_for_publish merges AST findings into the review when the LLM pass is a
no-op, and reports a clean verdict for a benign app.
no-op, reports a clean verdict for a benign app, and memoizes by source so an
unchanged reopen never re-bills the aux model.
Run with: backend/.venv/bin/python backend/tests/test_publish.py
"""
@@ -24,27 +25,27 @@ import tempfile
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
from backend.apps.outputs.models import Output
from backend.apps.outputs import publish
from backend.apps.outputs import publish_common, publish_scan, publish_build
def test_slugify():
assert publish.slugify("My Cool App!!") == "my-cool-app"
assert publish.slugify(" ") == "app"
assert publish.slugify("") == "app"
assert publish.slugify("a" * 100) == "a" * 32
assert publish.slugify("Café ☕ Menu") == "caf-menu"
assert publish_common.slugify("My Cool App!!") == "my-cool-app"
assert publish_common.slugify(" ") == "app"
assert publish_common.slugify("") == "app"
assert publish_common.slugify("a" * 100) == "a" * 32
assert publish_common.slugify("Café ☕ Menu") == "caf-menu"
def test_ast_gate_flags_unsafe_and_clean():
unsafe = Output(name="x", files={"backend.py": "import os\nresult={'c': os.getcwd()}\n"})
findings = publish.quick_ast_gate(unsafe)
findings = publish_scan.quick_ast_gate(unsafe)
assert findings and any("os" in f for f in findings)
clean = Output(name="x", files={"backend.py": "import math\nresult={'p': math.pi}\n"})
assert publish.quick_ast_gate(clean) == []
assert publish_scan.quick_ast_gate(clean) == []
no_backend = Output(name="x", files={"index.html": "<html>hi</html>"})
assert publish.quick_ast_gate(no_backend) == []
assert publish_scan.quick_ast_gate(no_backend) == []
def test_collect_source_filters():
@@ -54,7 +55,7 @@ def test_collect_source_filters():
"data.bin": "not source",
"notes.txt": "ignore me",
})
src = publish._collect_source(o)
src = publish_scan._collect_source(o)
assert set(src.keys()) == {"index.html", "backend.py"}
@@ -63,13 +64,30 @@ def test_collect_bundle_flat():
"index.html": "<html>hi</html>",
"backend.py": "result={}",
})
blob = publish.collect_bundle(o, None)
blob = publish_build.collect_bundle(o, None)
with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as t:
assert sorted(t.getnames()) == ["backend.py", "index.html"]
idx = t.extractfile("index.html").read().decode()
assert idx == "<html>hi</html>"
def test_collect_bundle_drops_secret_files():
# A public bundle must never carry secrets, in either mode.
o = Output(name="x", files={
"index.html": "<html>hi</html>",
".env": "OPENAI_API_KEY=sk-secret",
".env.local": "X=1",
"server.pem": "-----BEGIN PRIVATE KEY-----",
".npmrc": "//registry/:_authToken=abc",
"app.js": "console.log(1)",
})
blob = publish_build.collect_bundle(o, None)
with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as t:
names = set(t.getnames())
assert names == {"index.html", "app.js"}
assert not (names & {".env", ".env.local", "server.pem", ".npmrc"})
def test_collect_bundle_webapp_dist_skips_symlink():
o = Output(name="x", workspace_id="ws123")
with tempfile.TemporaryDirectory() as dist:
@@ -78,36 +96,67 @@ def test_collect_bundle_webapp_dist_skips_symlink():
f.write("<html>built</html>")
with open(os.path.join(dist, "assets", "app.js"), "w") as f:
f.write("console.log(1)")
with open(os.path.join(dist, ".env"), "w") as f:
f.write("SECRET=1")
try:
os.symlink(os.path.join(dist, "index.html"), os.path.join(dist, "link.html"))
except OSError:
pass
blob = publish.collect_bundle(o, dist)
blob = publish_build.collect_bundle(o, dist)
with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as t:
names = sorted(t.getnames())
assert "index.html" in names
assert "assets/app.js" in names
assert "link.html" not in names # symlinks are skipped
assert ".env" not in names # secrets are dropped
def test_scan_for_publish_merges_ast():
# Force the LLM pass to a deterministic no-op so the test is hermetic.
async def _no_llm(src, settings):
return [], "clean"
orig = publish._llm_findings
publish._llm_findings = _no_llm
orig = publish_scan._llm_findings
publish_scan._llm_findings = _no_llm
publish_scan._memo.clear()
try:
unsafe = Output(name="x", files={"backend.py": "import socket\nresult={}\n"})
review = asyncio.run(publish.scan_for_publish(unsafe, settings=object()))
review = asyncio.run(publish_scan.scan_for_publish(unsafe, settings=object()))
assert review.verdict == "warn"
assert any("socket" in f for f in review.findings)
clean = Output(name="x", files={"index.html": "<html>hi</html>"})
review2 = asyncio.run(publish.scan_for_publish(clean, settings=object()))
review2 = asyncio.run(publish_scan.scan_for_publish(clean, settings=object()))
assert review2.verdict == "clean"
assert review2.findings == []
finally:
publish._llm_findings = orig
publish_scan._llm_findings = orig
publish_scan._memo.clear()
def test_scan_memo_skips_second_llm_call():
# Unchanged source must not re-invoke the (paid) LLM pass on a reopen.
calls = {"n": 0}
async def _counting_llm(src, settings):
calls["n"] += 1
return ["from the llm"], "warn"
orig = publish_scan._llm_findings
publish_scan._llm_findings = _counting_llm
publish_scan._memo.clear()
try:
app = Output(name="x", files={"index.html": "<html>same</html>"})
r1 = asyncio.run(publish_scan.scan_for_publish(app, settings=object()))
r2 = asyncio.run(publish_scan.scan_for_publish(app, settings=object()))
assert calls["n"] == 1, "second scan of identical source should hit the memo"
assert r1.findings == r2.findings
changed = Output(name="x", files={"index.html": "<html>different</html>"})
asyncio.run(publish_scan.scan_for_publish(changed, settings=object()))
assert calls["n"] == 2, "changed source must bust the memo"
finally:
publish_scan._llm_findings = orig
publish_scan._memo.clear()
def test_runtime_injection():