Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 45 additions & 1 deletion backend/app/api/share.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,14 @@
import uuid

from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import PlainTextResponse
from fastapi.responses import HTMLResponse, PlainTextResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import CurrentUser, get_current_user
from app.db import get_read_session, get_session
from app.ddl.export import snapshot_json_to_sql
from app.spec.schema_docs_html import render_schema_docs_html
from app.models import (
ProjectMember,
SchemaSnapshot,
Expand Down Expand Up @@ -235,3 +236,46 @@ async def export_shared_snapshot_index_design(
status_code=502, detail="LLM provider request failed"
) from exc
return generate_index_design_spec(data.snapshot_json, mode=mode)


@router.get(
"/share/{share_link_uuid}/snapshots/{schema_snapshot_uuid}/docs",
response_class=HTMLResponse,
)
async def get_shared_snapshot_docs(
share_link_uuid: uuid.UUID,
schema_snapshot_uuid: uuid.UUID,
session: AsyncSession = Depends(get_read_session),
) -> HTMLResponse:
"""Public, read-only schema documentation page for a shared snapshot.

Same validation as the shared-snapshot JSON endpoint (link exists, not
expired, snapshot belongs to the link's project). Every rendered value is
HTML-escaped and the page is self-contained (no scripts), so a strict CSP
is applied.
"""
link = await session.get(ShareLink, share_link_uuid)
if link is None:
raise HTTPException(status_code=404, detail="share link not found")
if link.expires_at is not None and link.expires_at <= dt.datetime.now(
dt.timezone.utc
):
raise HTTPException(status_code=410, detail="share link expired")

snap = await session.get(SchemaSnapshot, schema_snapshot_uuid)
if snap is None or snap.project_space_uuid != link.project_space_uuid:
raise HTTPException(status_code=404, detail="snapshot not found")

data = await session.get(SchemaSnapshotData, schema_snapshot_uuid)
html_page = render_schema_docs_html(
data.snapshot_json if data else None, title="Schema documentation"
)
return HTMLResponse(
content=html_page,
headers={
"Content-Security-Policy": (
"default-src 'none'; style-src 'unsafe-inline'; img-src data:"
),
"X-Content-Type-Options": "nosniff",
},
)
101 changes: 101 additions & 0 deletions backend/app/spec/schema_docs_html.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
"""Render a snapshot as a self-contained public schema-docs HTML page.

The dbdocs-style deliverable: a share-link URL anyone can open and read —
tables, columns, keys, relationships — without an account. Security posture:

* every value from the snapshot is HTML-escaped (database identifiers and
comments are untrusted input);
* the page is fully self-contained (inline CSS, no scripts, no external
requests), so a strict CSP can be applied by the caller.
"""

from __future__ import annotations

import html
from typing import Any

_CSS = """
body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;
margin:2rem auto;max-width:60rem;padding:0 1rem;color:#1a202c;background:#fff}
h1{font-size:1.5rem;border-bottom:2px solid #e2e8f0;padding-bottom:.5rem}
h2{font-size:1.15rem;margin-top:2rem}
table{border-collapse:collapse;width:100%;font-size:.875rem}
th,td{border:1px solid #e2e8f0;padding:.35rem .6rem;text-align:left}
th{background:#f7fafc}
.badge{display:inline-block;font-size:.7rem;font-weight:600;padding:.1rem .4rem;
border-radius:.25rem;margin-left:.3rem}
.pk{background:#fefcbf;color:#744210}.fk{background:#bee3f8;color:#2a4365}
.nn{background:#e2e8f0;color:#2d3748}
.comment{color:#718096}
footer{margin-top:3rem;font-size:.75rem;color:#a0aec0}
""".strip()


def _esc(value: object) -> str:
return html.escape(str(value)) if value is not None else ""


def render_schema_docs_html(snapshot: dict[str, Any] | None, title: str = "Schema") -> str:
"""Render the snapshot as a static, escaped, self-contained HTML document."""
snapshot = snapshot or {}
relations = sorted(
(r for r in snapshot.get("relations") or []),
key=lambda r: (str(r.get("schema_name")), str(r.get("relation_name"))),
)
cols_by_oid: dict[Any, list[dict[str, Any]]] = {}
for c in snapshot.get("columns") or []:
cols_by_oid.setdefault(c.get("relation_oid"), []).append(c)
for cols in cols_by_oid.values():
cols.sort(key=lambda c: c.get("column_position") or 0)
pk_by_oid: dict[Any, set[str]] = {}
for pk in snapshot.get("pk_columns") or []:
pk_by_oid.setdefault(pk.get("relation_oid"), set()).add(str(pk.get("column_name")))
fk_cols: dict[tuple[Any, str], str] = {}
rel_by_oid = {r.get("relation_oid"): r for r in relations}
for e in snapshot.get("fk_edges") or []:
parent = rel_by_oid.get(e.get("parent_relation_oid"))
if parent is not None:
fk_cols[(e.get("child_relation_oid"), str(e.get("child_column_name")))] = (
f"{parent.get('schema_name')}.{parent.get('relation_name')}."
f"{e.get('parent_column_name')}"
)

parts = [
"<!doctype html>",
'<html lang="en"><head><meta charset="utf-8">',
'<meta name="viewport" content="width=device-width, initial-scale=1">',
f"<title>{_esc(title)}</title>",
f"<style>{_CSS}</style></head><body>",
f"<h1>{_esc(title)}</h1>",
f"<p class='comment'>{len(relations)} relations · generated by pg-erd-cloud</p>",
]
for rel in relations:
oid = rel.get("relation_oid")
qname = f"{rel.get('schema_name')}.{rel.get('relation_name')}"
kind = " (view)" if str(rel.get("relation_kind") or "r") in ("v", "m") else ""
parts.append(f"<h2>{_esc(qname)}{_esc(kind)}</h2>")
if rel.get("relation_comment"):
parts.append(f"<p class='comment'>{_esc(rel['relation_comment'])}</p>")
parts.append(
"<table><tr><th>#</th><th>Column</th><th>Type</th><th>Keys</th>"
"<th>Comment</th></tr>"
)
for i, col in enumerate(cols_by_oid.get(oid, []), start=1):
name = str(col.get("column_name"))
badges = []
if name in pk_by_oid.get(oid, set()):
badges.append("<span class='badge pk'>PK</span>")
ref = fk_cols.get((oid, name))
if ref:
badges.append(f"<span class='badge fk'>FK → {_esc(ref)}</span>")
if col.get("is_not_null"):
badges.append("<span class='badge nn'>NN</span>")
parts.append(
f"<tr><td>{i}</td><td>{_esc(name)}</td>"
f"<td>{_esc(col.get('data_type'))}</td>"
f"<td>{''.join(badges)}</td>"
f"<td class='comment'>{_esc(col.get('column_comment') or '')}</td></tr>"
)
parts.append("</table>")
parts.append("<footer>Read-only schema documentation.</footer></body></html>")
return "\n".join(parts)
61 changes: 61 additions & 0 deletions backend/tests/test_schema_docs_html.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
from __future__ import annotations

from app.spec.schema_docs_html import render_schema_docs_html

SNAP = {
"relations": [
{"relation_oid": 1, "relation_kind": "r", "schema_name": "public", "relation_name": "member", "relation_comment": "회원"},
{"relation_oid": 2, "relation_kind": "v", "schema_name": "public", "relation_name": "v_report"},
],
"columns": [
{"relation_oid": 1, "column_name": "member_id", "column_position": 1, "data_type": "bigint", "is_not_null": True},
{"relation_oid": 1, "column_name": "email", "column_position": 2, "data_type": "text", "is_not_null": False, "column_comment": "로그인 이메일"},
],
"pk_columns": [{"relation_oid": 1, "column_name": "member_id"}],
"fk_edges": [],
}


def test_renders_tables_columns_badges_and_comments():
page = render_schema_docs_html(SNAP, title="My Schema")
assert "<title>My Schema</title>" in page
assert "public.member" in page
assert "회원" in page and "로그인 이메일" in page
assert "class='badge pk'" in page # member_id PK badge
assert "(view)" in page # v_report labelled


def test_fk_badge_points_to_parent():
snap = {**SNAP, "relations": SNAP["relations"] + [
{"relation_oid": 3, "relation_kind": "r", "schema_name": "public", "relation_name": "orders"},
], "columns": SNAP["columns"] + [
{"relation_oid": 3, "column_name": "member_id", "column_position": 1, "data_type": "bigint", "is_not_null": True},
], "fk_edges": [
{"child_relation_oid": 3, "parent_relation_oid": 1,
"child_column_name": "member_id", "parent_column_name": "member_id"},
]}
page = render_schema_docs_html(snap)
assert "FK → public.member.member_id" in page


def test_everything_is_escaped_no_script_injection():
hostile = {
"relations": [{"relation_oid": 1, "relation_kind": "r", "schema_name": "public",
"relation_name": "<script>alert(1)</script>",
"relation_comment": '<img src=x onerror=alert(2)>'}],
"columns": [{"relation_oid": 1, "column_name": '"><script>x</script>',
"column_position": 1, "data_type": "<b>text</b>", "is_not_null": False}],
"pk_columns": [], "fk_edges": [],
}
page = render_schema_docs_html(hostile)
assert "<script>" not in page
assert "onerror=" not in page or "&lt;img" in page # escaped form only
assert "&lt;script&gt;alert(1)&lt;/script&gt;" in page
# page itself contains no script tags at all
assert page.lower().count("<script") == 0


def test_empty_snapshot_still_valid_page():
page = render_schema_docs_html(None)
assert page.startswith("<!doctype html>")
assert "0 relations" in page