200 lines
9.6 KiB
Python
200 lines
9.6 KiB
Python
"""Weboberfläche im echten Chrome prüfen: CSP-Verstöße zählen, Bildschirmfotos machen.
|
|
|
|
Startet einen Hugin-Server mit Testdaten, meldet sich mit verschiedenen Keys an
|
|
und ruft jede Seite über das DevTools-Protokoll auf. Ein vor dem Laden
|
|
eingefügtes Skript (CDP, nicht von der CSP betroffen) zählt Verstöße.
|
|
|
|
uv run --with websockets --with httpx python scripts/csp/web_browser.py <ausgabeordner>
|
|
"""
|
|
|
|
import asyncio
|
|
import base64
|
|
import io
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import websockets
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
CHROME = os.environ.get("CHROME", "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome")
|
|
ADMIN = "web-test-admin"
|
|
COUNTER = """window.__csp = []; document.addEventListener('securitypolicyviolation',
|
|
e => window.__csp.push(e.violatedDirective + ' ' + (e.sample || '').slice(0, 40)));"""
|
|
|
|
|
|
def free_port() -> int:
|
|
with socket.socket() as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
return s.getsockname()[1]
|
|
|
|
|
|
def zip_folder(folder: Path) -> bytes:
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w") as z:
|
|
for d in sorted(folder.rglob("*")):
|
|
if d.is_file():
|
|
z.write(d, d.relative_to(folder).as_posix())
|
|
return buffer.getvalue()
|
|
|
|
|
|
def create_data(url: str) -> dict[str, str]:
|
|
a = {"Authorization": f"Bearer {ADMIN}"}
|
|
c = httpx.Client(base_url=url)
|
|
keys = {}
|
|
for tenant, project in [("plattform", "kern"), ("shop", "api")]:
|
|
c.post("/api/v1/admin/tenants", headers=a, json={"slug": tenant, "name": tenant.title()}).raise_for_status()
|
|
c.post(f"/api/v1/admin/tenants/{tenant}/projects", headers=a, json={"slug": project, "name": project})
|
|
for name, tenant, project, roles in [("Autorin", "plattform", "kern", ["pub_author"]),
|
|
("Prüfer", "plattform", "kern", ["pub_reviewer"]),
|
|
("Shop CI", "shop", "api", [])]:
|
|
response = c.post(f"/api/v1/admin/tenants/{tenant}/keys", headers=a,
|
|
json={"kind": "project", "name": name, "project": project, "roles": roles})
|
|
keys[name] = response.json()["key"]
|
|
packages = ROOT / "contract/examples/packages"
|
|
for name in ["pub-munin-nutzen/1.4.0", "pub-hausregeln/1.0.0", "pub-munin-anbindung/1.0.0",
|
|
"pub-munin-nutzen/1.5.0"]:
|
|
c.post("/api/v1/drafts", headers={"Authorization": f"Bearer {keys['Autorin']}",
|
|
"Content-Type": "application/zip"},
|
|
content=zip_folder(packages / name)).raise_for_status()
|
|
package, number = name.split("/")
|
|
base = f"/api/v1/packages/{package}/versions/{number}"
|
|
c.post(f"{base}/submit", headers={"Authorization": f"Bearer {keys['Autorin']}"}).raise_for_status()
|
|
if name != "pub-munin-nutzen/1.5.0": # 1.5.0 bleibt in Prüfung
|
|
c.post(f"{base}/approve", headers={"Authorization": f"Bearer {keys['Prüfer']}"},
|
|
json={"comment": "passt"}).raise_for_status()
|
|
c.post("/api/v1/usage", headers={"Authorization": f"Bearer {keys['Shop CI']}"},
|
|
json={"repo": "git.intern/shop/api", "tools": {"claude": "2.1.302"},
|
|
"installed": {"pub-munin-nutzen": "1.4.0", "pub-hausregeln": "1.0.0"}})
|
|
keys["Admin"] = ADMIN
|
|
return keys
|
|
|
|
|
|
def cookie_for(url: str, key: str) -> str:
|
|
response = httpx.post(f"{url}/login", data={"key": key, "next": ""})
|
|
return response.cookies["hugin_session"]
|
|
|
|
|
|
async def cdp(ws, method, **params):
|
|
cdp.n += 1
|
|
await ws.send(json.dumps({"id": cdp.n, "method": method, "params": params}))
|
|
while True:
|
|
response = json.loads(await ws.recv())
|
|
if response.get("id") == cdp.n:
|
|
return response.get("result", {})
|
|
|
|
|
|
cdp.n = 0
|
|
|
|
|
|
async def validate_pages(port_chrome: int, url: str, pages, output: Path, dunkel: bool) -> int:
|
|
target = httpx.put(f"http://127.0.0.1:{port_chrome}/json/new?about:blank").json()
|
|
error = 0
|
|
async with websockets.connect(target["webSocketDebuggerUrl"], max_size=50_000_000) as ws:
|
|
await cdp(ws, "Page.enable")
|
|
await cdp(ws, "Network.enable")
|
|
await cdp(ws, "Page.addScriptToEvaluateOnNewDocument", source=COUNTER)
|
|
await cdp(ws, "Emulation.setDeviceMetricsOverride", width=1280, height=900, deviceScaleFactor=1, mobile=False)
|
|
if dunkel:
|
|
await cdp(ws, "Emulation.setEmulatedMedia", features=[{"name": "prefers-color-scheme", "value": "dark"}])
|
|
for name, path, cookie in pages:
|
|
await cdp(ws, "Network.clearBrowserCookies")
|
|
if cookie:
|
|
await cdp(ws, "Network.setCookie", name="hugin_session", value=cookie, url=url)
|
|
await cdp(ws, "Page.navigate", url=url + path)
|
|
await asyncio.sleep(1.2)
|
|
result = await cdp(ws, "Runtime.evaluate", expression="JSON.stringify(window.__csp)", returnByValue=True)
|
|
verstoesse = json.loads(result["result"]["value"])
|
|
error += len(verstoesse)
|
|
hoehe = (await cdp(ws, "Runtime.evaluate", expression="document.documentElement.scrollHeight",
|
|
returnByValue=True))["result"]["value"]
|
|
image = await cdp(ws, "Page.captureScreenshot", format="png", captureBeyondViewport=True,
|
|
clip={"x": 0, "y": 0, "width": 1280, "height": min(hoehe, 4000), "scale": 1})
|
|
file = output / f"{'dunkel-' if dunkel else ''}{name}.png"
|
|
file.write_bytes(base64.b64decode(image["data"]))
|
|
print(f"{'✓' if not verstoesse else '✗'} {name:<12} {path:<45} Verstöße: {len(verstoesse)} {verstoesse[:3]}")
|
|
return error
|
|
|
|
|
|
async def gegenprobe(port_chrome: int, url: str, cookie: str) -> bool:
|
|
"""Fügt ein style-Attribut ein; die Messung muss den Verstoß sehen."""
|
|
target = httpx.put(f"http://127.0.0.1:{port_chrome}/json/new?about:blank").json()
|
|
async with websockets.connect(target["webSocketDebuggerUrl"]) as ws:
|
|
await cdp(ws, "Page.enable")
|
|
await cdp(ws, "Network.enable")
|
|
await cdp(ws, "Page.addScriptToEvaluateOnNewDocument", source=COUNTER)
|
|
await cdp(ws, "Network.setCookie", name="hugin_session", value=cookie, url=url)
|
|
await cdp(ws, "Page.navigate", url=url + "/help")
|
|
await asyncio.sleep(1)
|
|
await cdp(ws, "Runtime.evaluate", expression="document.body.insertAdjacentHTML('beforeend', "
|
|
"'<div style=\\'color:red\\'>x</div>')")
|
|
await asyncio.sleep(0.5)
|
|
anzahl = (await cdp(ws, "Runtime.evaluate", expression="window.__csp.length", returnByValue=True))
|
|
return anzahl["result"]["value"] > 0
|
|
|
|
|
|
def main() -> None:
|
|
output = Path(sys.argv[1])
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
arbeit = Path(tempfile.mkdtemp())
|
|
port, port_chrome = free_port(), free_port()
|
|
url = f"http://127.0.0.1:{port}"
|
|
env = os.environ | {"HUGIN_DATABASE_URL": f"sqlite:///{arbeit}/hugin.db", "HUGIN_ADMIN_TOKEN": ADMIN}
|
|
server_folder = ROOT / "server"
|
|
subprocess.run(["uv", "run", "-q", "hugin-server", "migrate"], cwd=server_folder, env=env, check=True)
|
|
server = subprocess.Popen(["uv", "run", "-q", "hugin-server", "run", "--host", "127.0.0.1", "--port", str(port)],
|
|
cwd=server_folder, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
chrome = subprocess.Popen([CHROME, "--headless=new", "--disable-gpu", f"--remote-debugging-port={port_chrome}",
|
|
f"--user-data-dir={arbeit}/chrome", "about:blank"],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
for _ in range(100):
|
|
try:
|
|
httpx.get(f"{url}/readyz").raise_for_status()
|
|
httpx.get(f"http://127.0.0.1:{port_chrome}/json/version").raise_for_status()
|
|
break
|
|
except httpx.HTTPError:
|
|
time.sleep(0.2)
|
|
keys = create_data(url)
|
|
c = {name: cookie_for(url, key) for name, key in keys.items()}
|
|
pages = [
|
|
("login", "/login", None),
|
|
("catalog", "/catalog", c["Shop CI"]),
|
|
("package", "/packages/pub-munin-nutzen", c["Shop CI"]),
|
|
("version", "/packages/pub-munin-nutzen/versions/1.4.0", c["Shop CI"]),
|
|
("pruefung", "/packages/pub-munin-nutzen/versions/1.5.0", c["Prüfer"]),
|
|
("reviews", "/reviews", c["Prüfer"]),
|
|
("upload", "/upload", c["Autorin"]),
|
|
("project", "/project", c["Shop CI"]),
|
|
("manage", "/manage?tenant=shop", c["Admin"]),
|
|
("overview", "/overview", c["Shop CI"]),
|
|
("system", "/system", c["Admin"]),
|
|
("audit", "/audit", c["Admin"]),
|
|
("new", "/new?kind=skill&id=pub-neu", c["Autorin"]),
|
|
("editor", "/packages/pub-munin-nutzen/versions/1.4.0/edit", c["Autorin"]),
|
|
("download", "/download", c["Shop CI"]),
|
|
("help", "/help/package-format", c["Shop CI"]),
|
|
("hilfe-ohne", "/help/teams", None),
|
|
]
|
|
if not asyncio.run(gegenprobe(port_chrome, url, c["Shop CI"])):
|
|
print("Gegenprobe gescheitert: ein absichtlicher Verstoß wurde nicht gezählt")
|
|
sys.exit(2)
|
|
print("Gegenprobe ok: absichtlicher Verstoß wird gezählt")
|
|
error = asyncio.run(validate_pages(port_chrome, url, pages, output, dunkel=False))
|
|
error += asyncio.run(validate_pages(port_chrome, url, pages[1:4], output, dunkel=True))
|
|
print("Verstöße gesamt:", error)
|
|
sys.exit(1 if error else 0)
|
|
finally:
|
|
server.terminate()
|
|
chrome.terminate()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|