--

src/nercone_website/server.py

@@ -18,9 +18,8 @@ from jinja2.exceptions import TemplateNotFound
 from .error import error_page
 from .database import AccessCounter
 from .middleware import Middleware, server_version, onion_hostname
-from .tools.tls_test import TlsJobQueue, TlsTestDB
-from .tools.tls_test.engine import run_full_scan, parse_target
-from .tools.tls_test.ratelimit import check as ratelimit_check, client_ip_from_scope
+from .tools.tls_test import TlsJobQueue, TlsTestDB, tls_submit, tls_api_submit, tls_results_context
+from .tools.tls_test.engine import run_full_scan
 
 tls_test_db = TlsTestDB()
 tls_test_queue = TlsJobQueue(tls_test_db, run_full_scan)
@@ -144,39 +143,6 @@ async def thumbnail(request: Request, path: str) -> Response:
     png = resvg_py.svg_to_bytes(svg, font_files=font_files, width=1200, height=630)
     return Response(content=png, media_type="image/png")
 
-def _validate_tls_target(raw: str) -> str | None:
-    import ipaddress
-    s = (raw or "").strip()
-    if not s or len(s) > 255:
-        return None
-    # Reject obviously garbage input (all dashes/dots, control chars, whitespace).
-    if not re.compile(r"^[A-Za-z0-9._:\[\]\-]{1,255}$").match(s):
-        return None
-    try:
-        host, port, _ = parse_target(s)
-    except Exception:
-        return None
-    if not host or port <= 0 or port > 65535:
-        return None
-    # Host must be either an IP literal or a valid-looking hostname.
-    try:
-        ipaddress.ip_address(host)
-        return s
-    except ValueError:
-        pass
-    # Hostname: each label must be 1-63 chars, not start/end with '-', no empty labels.
-    labels = host.split(".")
-    if not labels or any(not lbl for lbl in labels):
-        return None
-    for lbl in labels:
-        if len(lbl) > 63 or lbl.startswith("-") or lbl.endswith("-"):
-            return None
-        if not re.match(r"^[A-Za-z0-9\-]+$", lbl):
-            return None
-    if len(labels) < 2 and host != "localhost":
-        return None
-    return s
-
 @app.api_route("/tools/tls-test/", methods=["GET"])
 async def tls_test_index(request: Request) -> Response:
     return templates.TemplateResponse(request=request, name="tools/tls-test/index.html")
@@ -184,25 +150,9 @@ async def tls_test_index(request: Request) -> Response:
 @app.api_route("/tools/tls-test/", methods=["POST"])
 async def tls_test_submit(request: Request) -> Response:
     form = await request.form()
-    raw = str(form.get("target", "")).strip()
-    target = _validate_tls_target(raw)
-    if not target:
-        return templates.TemplateResponse(
-            request=request,
-            name="tools/tls-test/index.html",
-            context={"error": "無効なターゲットです。ホスト名/IP(:ポート)を入力してください。", "last_target": raw},
-            status_code=400,
-        )
-    client_ip = client_ip_from_scope(request.scope)
-    decision = ratelimit_check(tls_test_db, client_ip)
-    if not decision.allowed:
-        return templates.TemplateResponse(
-            request=request,
-            name="tools/tls-test/index.html",
-            context={"error": decision.reason, "last_target": raw},
-            status_code=429,
-        )
-    test_id = tls_test_queue.submit(target, client_ip)
+    test_id, err = tls_submit(request, str(form.get("target", "")).strip(), tls_test_db, tls_test_queue, templates)
+    if err:
+        return err
     return RedirectResponse(url=f"/tools/tls-test/status/{test_id}/", status_code=303)
 
 @app.api_route("/tools/tls-test/status/{test_id}/", methods=["GET"])
@@ -212,16 +162,10 @@ async def tls_test_status_page(request: Request, test_id: str) -> Response:
         return error_page(templates, request, 404, "指定されたテストが見つかりません。", "…id間違ってない？")
     if job.get("status") == "done":
         return RedirectResponse(url=f"/tools/tls-test/results/{test_id}/", status_code=303)
-    progress = tls_test_db.get_progress(test_id)
     return templates.TemplateResponse(
         request=request,
         name="tools/tls-test/status.html",
-        context={
-            "test_id": test_id,
-            "target": job.get("target", ""),
-            "status": job.get("status", ""),
-            "progress_entries": progress,
-        },
+        context={"test_id": test_id, "target": job.get("target", ""), "status": job.get("status", ""), "progress_entries": tls_test_db.get_progress(test_id)},
     )
 
 @app.api_route("/tools/tls-test/results/{test_id}/", methods=["GET"])
@@ -231,96 +175,7 @@ async def tls_test_results_page(request: Request, test_id: str) -> Response:
         return error_page(templates, request, 404, "指定されたテストが見つかりません。", "…id間違ってない？")
     if job.get("status") != "done":
         return RedirectResponse(url=f"/tools/tls-test/status/{test_id}/", status_code=303)
-    result = job.get("result") or {}
-    findings = result.get("findings", [])
-    # Primary grouping — the 4 tabs requested by the user.
-    groups: dict[str, list[dict]] = {
-        "reliability": [],
-        "safety": [],
-        "vulnerabilities": [],
-        "compatibility": [],
-        "other": [],
-    }
-    for f in findings:
-        g = f.get("group") or "other"
-        groups.setdefault(g, []).append(f)
-    # Summary tab — score-impacting findings, sorted by impact desc.
-    impactful = sorted(
-        [f for f in findings if (f.get("impact") or 0) > 0],
-        key=lambda f: f.get("impact", 0),
-        reverse=True,
-    )
-    # Always include the top "good" findings too so the Summary tab is not empty
-    # on pristine sites. Take up to 20 impactful + 6 positives.
-    positives = [f for f in findings if f.get("severity") == "good"][:6]
-    summary = impactful[:20] + positives
-
-    # Human-friendly "finished_at" for the results header: ISO-like local string.
-    import datetime as _dt
-    finished_ts = job.get("finished_at") or job.get("started_at") or 0
-    finished_at_display = ""
-    try:
-        if finished_ts:
-            finished_at_display = _dt.datetime.fromtimestamp(int(finished_ts)).strftime("%Y-%m-%dT%H:%M:%S")
-    except Exception:
-        finished_at_display = ""
-
-    # Sharable canonical URL for the "Copy link" button. Prefer the public
-    # tools.* subdomain when known, otherwise the current request's origin.
-    share_url = str(request.url).split("#", 1)[0]
-
-    # ---- Reconstruct the "live log" that was shown on the processing page ----
-    # Progress entries are phase-transition messages stored in test_progress
-    # (ordered by seq). Findings arrive between progress rows and are grouped
-    # by the phase they ran under (Finding.step). We interleave the two so the
-    # ログ tab reads exactly like the real-time WS stream did.
-    progress_entries = tls_test_db.get_progress(test_id)
-    findings_by_step: dict[str, list[dict]] = {}
-    for f in findings:
-        step = (f.get("step") or "").strip()
-        findings_by_step.setdefault(step, []).append(f)
-    log_entries: list[dict] = []
-    seen_steps: set[str] = set()
-    for p in progress_entries:
-        log_entries.append({
-            "kind": "phase",
-            "phase": p.get("phase") or "",
-            "detail": p.get("detail") or "",
-            "severity": p.get("severity") or "info",
-        })
-        step = (p.get("phase") or "").strip()
-        if step and step not in seen_steps:
-            for f in findings_by_step.get(step, []):
-                log_entries.append({"kind": "finding", "finding": f})
-            seen_steps.add(step)
-    # Any findings whose step never showed up as a progress row (rare — engine
-    # errors etc.) — append them at the end so nothing is dropped.
-    for step, fs in findings_by_step.items():
-        if step not in seen_steps:
-            for f in fs:
-                log_entries.append({"kind": "finding", "finding": f})
-
-    return templates.TemplateResponse(
-        request=request,
-        name="tools/tls-test/results.html",
-        context={
-            "test_id": test_id,
-            "job": job,
-            "result": result,
-            "groups": groups,
-            "summary": summary,
-            "group_labels": {
-                "reliability": "信頼性",
-                "safety": "安全性",
-                "vulnerabilities": "脆弱性",
-                "compatibility": "互換性",
-                "other": "その他",
-            },
-            "finished_at_display": finished_at_display,
-            "share_url": share_url,
-            "log_entries": log_entries,
-        },
-    )
+    return templates.TemplateResponse(request=request, name="tools/tls-test/results.html", context=tls_results_context(job, test_id, request, tls_test_db))
 
 @app.websocket("/tools/tls-test/ws/{test_id}")
 async def tls_test_ws(websocket: WebSocket, test_id: str):
@@ -365,14 +220,9 @@ async def tls_test_api_scan(request: Request) -> Response:
         payload = await request.json()
     except Exception:
         return JSONResponse({"error": "invalid JSON"}, status_code=400)
-    target = _validate_tls_target(str(payload.get("target", "")))
-    if not target:
-        return JSONResponse({"error": "invalid target"}, status_code=400)
-    client_ip = client_ip_from_scope(request.scope)
-    decision = ratelimit_check(tls_test_db, client_ip)
-    if not decision.allowed:
-        return JSONResponse({"error": decision.reason}, status_code=429)
-    test_id = tls_test_queue.submit(target, client_ip)
+    test_id, err = tls_api_submit(request, str(payload.get("target", "")), tls_test_db, tls_test_queue)
+    if err:
+        return JSONResponse(err[0], status_code=err[1])
     return JSONResponse({
         "id": test_id,
         "status_url": f"/tools/tls-test/status/{test_id}/",

src/nercone_website/tools/tls_test/__init__.py

@@ -1,4 +1,5 @@
 from .runner import TlsJobQueue
 from .db import TlsTestDB
+from .views import tls_submit, tls_api_submit, tls_results_context
 
-__all__ = ["TlsJobQueue", "TlsTestDB"]
+__all__ = ["TlsJobQueue", "TlsTestDB", "tls_submit", "tls_api_submit", "tls_results_context"]

src/nercone_website/tools/tls_test/engine.py

@@ -75,6 +75,37 @@ def parse_target(target: str) -> tuple[str, int, str | None]:
     return host, port, sni
 
 
+def validate_tls_target(raw: str) -> str | None:
+    import re
+    s = (raw or "").strip()
+    if not s or len(s) > 255:
+        return None
+    if not re.compile(r"^[A-Za-z0-9._:\[\]\-]{1,255}$").match(s):
+        return None
+    try:
+        host, port, _ = parse_target(s)
+    except Exception:
+        return None
+    if not host or port <= 0 or port > 65535:
+        return None
+    try:
+        ipaddress.ip_address(host)
+        return s
+    except ValueError:
+        pass
+    labels = host.split(".")
+    if not labels or any(not lbl for lbl in labels):
+        return None
+    for lbl in labels:
+        if len(lbl) > 63 or lbl.startswith("-") or lbl.endswith("-"):
+            return None
+        if not re.match(r"^[A-Za-z0-9\-]+$", lbl):
+            return None
+    if len(labels) < 2 and host != "localhost":
+        return None
+    return s
+
+
 async def _gather(report: ReportProgress, finds: ReportFinding, result: ScanResult) -> None:
     host = result.host
     port = result.port

src/nercone_website/tools/tls_test/views.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+import datetime
+from fastapi import Request
+from fastapi.templating import Jinja2Templates
+from fastapi.responses import Response
+from .engine import validate_tls_target
+from .ratelimit import check as ratelimit_check, client_ip_from_scope
+
+
+def tls_submit(
+    request: Request,
+    raw: str,
+    tls_test_db,
+    tls_test_queue,
+    templates: Jinja2Templates,
+) -> tuple[str | None, Response | None]:
+    target = validate_tls_target(raw)
+    if not target:
+        return None, templates.TemplateResponse(
+            request=request,
+            name="tools/tls-test/index.html",
+            context={"error": "無効なターゲットです。ホスト名/IP(:ポート)を入力してください。", "last_target": raw},
+            status_code=400,
+        )
+    client_ip = client_ip_from_scope(request.scope)
+    decision = ratelimit_check(tls_test_db, client_ip)
+    if not decision.allowed:
+        return None, templates.TemplateResponse(
+            request=request,
+            name="tools/tls-test/index.html",
+            context={"error": decision.reason, "last_target": raw},
+            status_code=429,
+        )
+    return tls_test_queue.submit(target, client_ip), None
+
+
+def tls_api_submit(request: Request, raw: str, tls_test_db, tls_test_queue) -> tuple[str | None, tuple[dict, int] | None]:
+    target = validate_tls_target(raw)
+    if not target:
+        return None, ({"error": "invalid target"}, 400)
+    client_ip = client_ip_from_scope(request.scope)
+    decision = ratelimit_check(tls_test_db, client_ip)
+    if not decision.allowed:
+        return None, ({"error": decision.reason}, 429)
+    return tls_test_queue.submit(target, client_ip), None
+
+
+def tls_results_context(job: dict, test_id: str, request: Request, tls_test_db) -> dict:
+    result = job.get("result") or {}
+    findings = result.get("findings", [])
+
+    groups: dict[str, list[dict]] = {"reliability": [], "safety": [], "vulnerabilities": [], "compatibility": [], "other": []}
+    for f in findings:
+        groups.setdefault(f.get("group") or "other", []).append(f)
+
+    impactful = sorted([f for f in findings if (f.get("impact") or 0) > 0], key=lambda f: f.get("impact", 0), reverse=True)
+    summary = impactful[:20] + [f for f in findings if f.get("severity") == "good"][:6]
+
+    finished_at_display = ""
+    try:
+        ts = job.get("finished_at") or job.get("started_at") or 0
+        if ts:
+            finished_at_display = datetime.datetime.fromtimestamp(int(ts)).strftime("%Y-%m-%dT%H:%M:%S")
+    except Exception:
+        pass
+
+    progress_entries = tls_test_db.get_progress(test_id)
+    findings_by_step: dict[str, list[dict]] = {}
+    for f in findings:
+        findings_by_step.setdefault((f.get("step") or "").strip(), []).append(f)
+    log_entries: list[dict] = []
+    seen_steps: set[str] = set()
+    for p in progress_entries:
+        log_entries.append({"kind": "phase", "phase": p.get("phase") or "", "detail": p.get("detail") or "", "severity": p.get("severity") or "info"})
+        step = (p.get("phase") or "").strip()
+        if step and step not in seen_steps:
+            log_entries += [{"kind": "finding", "finding": f} for f in findings_by_step.get(step, [])]
+            seen_steps.add(step)
+    for step, fs in findings_by_step.items():
+        if step not in seen_steps:
+            log_entries += [{"kind": "finding", "finding": f} for f in fs]
+
+    return {
+        "test_id": test_id,
+        "job": job,
+        "result": result,
+        "groups": groups,
+        "summary": summary,
+        "group_labels": {"reliability": "信頼性", "safety": "安全性", "vulnerabilities": "脆弱性", "compatibility": "互換性", "other": "その他"},
+        "finished_at_display": finished_at_display,
+        "share_url": str(request.url).split("#", 1)[0],
+        "log_entries": log_entries,
+    }