From 73680d60b69d84a3e26bac24a66f4ed85e65b7a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 25 Jun 2026 18:04:29 +0000 Subject: [PATCH] Add optional local web UI for analysis, downloads, and run diffs (#23) Add a thin, optional Flask-based web layer over the existing analysis engine so domain lists can be analyzed from the browser without the command line. Web UI (domain_security_analyzer/web/, installed via the new [web] extra): - Upload domains by pasting (one per line, '#' comments ignored) or dropping a .txt file. - Analysis runs in a background thread with a live progress bar (polled via a JSON status endpoint); the page stays responsive. - Download the standard 29-column CSV for any run. - Changes view diffs the two most recent runs and classifies per-domain field deltas into security regressions (e.g. SPF/DMARC lost, SRI coverage dropped) vs. improvements, plus domains added/removed. - Launched with the new `domain-analyzer-web` console command (or `python -m domain_security_analyzer.web`); binds to 127.0.0.1 only and stores runs under ~/.domain-security-analyzer/runs/ (override via DSA_DATA_DIR). Engine refactor (backward compatible): - Extract the inline CSV writing in analyze_domains_from_file into a reusable write_results_csv() helper plus a shared CSV_COLUMNS constant. - Add an optional progress_callback(completed, total) parameter so the web UI can drive its progress bar without scraping stdout. Packaging / CI: - New optional dependency group [web] = flask>=3.0; flask also added to [dev]. - Register the domain-analyzer-web entry point and package the web templates and static assets in both wheel and sdist. - CI installs the [web] extra and smoke-tests the new entry point. Tests: add tests/test_web.py (network-free) covering domain parsing, the run-diff classification, and the full HTTP flow (run -> progress -> results -> download) with the engine monkeypatched, plus path-traversal rejection. Full suite: 45 passing. Implements the MVP scope of #23; cross-run history beyond the last two runs (SQLite) remains out of scope there. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01XYvSCw1sjoG2AHbLPCsX7S --- .github/workflows/ci.yml | 3 +- CHANGELOG.md | 20 ++ MANIFEST.in | 2 + README.md | 26 +++ domain_security_analyzer/analyzer.py | 167 ++++++++------ domain_security_analyzer/web/__init__.py | 17 ++ domain_security_analyzer/web/__main__.py | 5 + domain_security_analyzer/web/app.py | 205 ++++++++++++++++++ domain_security_analyzer/web/cli.py | 50 +++++ domain_security_analyzer/web/runs.py | 143 ++++++++++++ domain_security_analyzer/web/static/style.css | 104 +++++++++ .../web/templates/base.html | 24 ++ .../web/templates/changes.html | 50 +++++ .../web/templates/index.html | 41 ++++ .../web/templates/progress.html | 42 ++++ .../web/templates/results.html | 27 +++ pyproject.toml | 9 + tests/test_web.py | 154 +++++++++++++ 18 files changed, 1017 insertions(+), 72 deletions(-) create mode 100644 domain_security_analyzer/web/__init__.py create mode 100644 domain_security_analyzer/web/__main__.py create mode 100644 domain_security_analyzer/web/app.py create mode 100644 domain_security_analyzer/web/cli.py create mode 100644 domain_security_analyzer/web/runs.py create mode 100644 domain_security_analyzer/web/static/style.css create mode 100644 domain_security_analyzer/web/templates/base.html create mode 100644 domain_security_analyzer/web/templates/changes.html create mode 100644 domain_security_analyzer/web/templates/index.html create mode 100644 domain_security_analyzer/web/templates/progress.html create mode 100644 domain_security_analyzer/web/templates/results.html create mode 100644 tests/test_web.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4987d5d..f4a2657 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,7 +25,7 @@ jobs: - name: Install package and test dependencies run: | python -m pip install --upgrade pip - pip install -e . + pip install -e ".[web]" pip install pytest - name: Run tests run: pytest -q @@ -34,6 +34,7 @@ jobs: domain-analyzer --version domain-analyzer --help python -m domain_security_analyzer --version + domain-analyzer-web --help build: name: build & metadata check diff --git a/CHANGELOG.md b/CHANGELOG.md index d9d0db5..09bdf4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **SRI library API**: Subresource Integrity analysis is now importable from the + installed package (`from domain_security_analyzer.sri import scan_url`), + returning a structured report with per-resource integrity/crossorigin values, + machine-readable reason codes, observed CSP headers, and compensating-control + detection. `scripts/sri_parser.py` is now a thin CLI wrapper over it. +- **Local web UI** (optional `web` extra): `pip install domain-security-analyzer[web]` + adds a `domain-analyzer-web` command that serves a localhost Flask app to + upload domain lists, watch analysis progress, download the standard 29-column + CSV, and diff the two most recent runs (highlighting security regressions vs. + improvements). Runs are stored under `~/.domain-security-analyzer/runs/` + (override via `DSA_DATA_DIR`). + +### Changed + +- `analyzer.py` exposes a reusable `write_results_csv()` helper, a shared + `CSV_COLUMNS` constant, and an optional `progress_callback` on + `analyze_domains_from_file` (used by the web UI). Backward compatible. + ## [1.0.0] - 2026-06-21 First packaged release, distributable via PyPI. diff --git a/MANIFEST.in b/MANIFEST.in index 356ccf7..2223ed7 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -7,3 +7,5 @@ include requirements.txt recursive-include docs *.md recursive-include examples * include scripts/*.py +recursive-include domain_security_analyzer/web/templates *.html +recursive-include domain_security_analyzer/web/static *.css diff --git a/README.md b/README.md index 186b5a8..9bd6c72 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,32 @@ python -m domain_security_analyzer examples/domains.txt report.csv python domain_analyzer.py examples/domains.txt report.csv # legacy shim ``` +### Web UI (local app) + +Prefer a browser? Install the optional `web` extra and launch a small local app +to upload domains, watch progress, download the report, and compare runs — no +command line needed after launch: + +```bash +pip install "domain-security-analyzer[web]" +domain-analyzer-web --open # serves http://127.0.0.1:8000 and opens it +``` + +The app lets you: + +- **Upload** domains by pasting them (one per line, `#` comments ignored) or + dropping a `.txt` file. +- **Watch progress** as the analysis runs in the background. +- **Download** the standard 29-column CSV for any run. +- **View changes** — each run is saved locally and the *Changes* page diffs the + two most recent runs, highlighting security **regressions** (e.g. SPF/DMARC + lost, SRI coverage dropped) and **improvements**, plus domains added/removed. + +It binds to `127.0.0.1` only and stores runs under +`~/.domain-security-analyzer/runs/` (override with the `DSA_DATA_DIR` +environment variable). This is a single-user local tool — it is not meant to be +exposed to a network. + The generated CSV includes comprehensive security analysis with **29 columns**: ### **Domain & Infrastructure** diff --git a/domain_security_analyzer/analyzer.py b/domain_security_analyzer/analyzer.py index 29a7997..e2317a3 100644 --- a/domain_security_analyzer/analyzer.py +++ b/domain_security_analyzer/analyzer.py @@ -8,7 +8,7 @@ import concurrent.futures import csv from datetime import datetime -from typing import Dict, List, Optional +from typing import Callable, Dict, List, Optional from urllib.parse import urlparse import dns.resolver @@ -401,8 +401,95 @@ def analyze_domain(self, domain: str) -> Dict: } -def analyze_domains_from_file(input_file: str, output_file: str, max_workers: int = 10, *, include_wildcard_matches: bool = False, filtered_subdomains_file: Optional[str] = None): - """Analyze multiple domains from a file and save results to CSV.""" +# Column order for the analysis report CSV. Kept as a module-level constant so +# consumers (CLI, web UI, diff tooling) share one source of truth. +CSV_COLUMNS = [ + 'Domain', + 'Timestamp', + 'Parent Domain', + 'SOA Exists', + 'SOA Record', + 'Primary NS', + 'Admin Email', + 'SPF Exists', + 'SPF Record', + 'DKIM Exists', + 'DKIM Records', + 'DMARC Exists', + 'DMARC Record', + 'Discovered Subdomains', + 'CNAME Records', + 'Has Wildcard DNS', + 'Hosting Provider', + 'HTTP Accessible', + 'Redirects to HTTPS', + 'Final URL', + 'Redirect Chain', + 'HTTP Error', + 'SRI Enabled', + 'Total External Resources', + 'Resources With SRI', + 'SRI Coverage %', + 'Missing SRI Count', + 'SRI Algorithms Used', + 'SRI Error', +] + + +def _result_to_row(r: Dict) -> list: + """Flatten one analysis result dict into a CSV row matching CSV_COLUMNS.""" + return [ + r['domain'], + r['timestamp'], + r['soa']['parent_domain'], + r['soa']['exists'], + r['soa'].get('record'), + r['soa'].get('primary_ns'), + r['soa'].get('admin_email'), + r['spf']['exists'], + r['spf'].get('record'), + r['dkim']['exists'], + ';'.join([f"{rec['selector']}:{rec['record']}" for rec in r['dkim']['records']]) if r['dkim']['records'] else '', + r['dmarc']['exists'], + r['dmarc'].get('record'), + ','.join(r['subdomains']['subdomains']), + ','.join([f"{k}:{v}" for k, v in r['subdomains']['cname_records'].items()]), + r['subdomains']['has_wildcard_dns'], + r['subdomains']['hosting_provider'], + r['http_redirect']['http_accessible'], + r['http_redirect']['redirects_to_https'], + r['http_redirect']['final_url'], + ' -> '.join(r['http_redirect'].get('redirect_chain', [])), + r['http_redirect']['error'], + r['sri']['sri_enabled'], + r['sri']['total_external_resources'], + r['sri']['resources_with_sri'], + r['sri']['sri_coverage_percentage'], + r['sri']['missing_sri_count'], + ','.join(r['sri']['sri_algorithms_used']) if r['sri']['sri_algorithms_used'] else '', + r['sri']['error'], + ] + + +def write_results_csv(results: List[Dict], output_file: str) -> None: + """Write analysis result dicts to ``output_file`` as the standard report CSV. + + Shared by the CLI and the web UI so the 29-column layout has a single + definition. + """ + with open(output_file, 'w', newline='') as f: + writer = csv.writer(f) + writer.writerow(CSV_COLUMNS) + for r in results: + writer.writerow(_result_to_row(r)) + + +def analyze_domains_from_file(input_file: str, output_file: str, max_workers: int = 10, *, include_wildcard_matches: bool = False, filtered_subdomains_file: Optional[str] = None, progress_callback: Optional[Callable[[int, int], None]] = None): + """Analyze multiple domains from a file and save results to CSV. + + ``progress_callback``, if given, is invoked as ``callback(completed, total)`` + after each domain finishes — used by the web UI to drive a progress bar. + """ # Read domains from input file with open(input_file, 'r') as f: @@ -434,6 +521,11 @@ def analyze_single_domain(domain: str) -> Dict: completed += 1 print(f"Progress: {completed}/{total_domains} domains analyzed ({(completed/total_domains)*100:.1f}%)") + if progress_callback is not None: + try: + progress_callback(completed, total_domains) + except Exception: + pass # progress reporting must never break analysis return result results = [] @@ -466,74 +558,7 @@ def analyze_single_domain(domain: str) -> Dict: results.append(error_result) # Write results to CSV - with open(output_file, 'w', newline='') as f: - writer = csv.writer(f) - # Write header - writer.writerow([ - 'Domain', - 'Timestamp', - 'Parent Domain', - 'SOA Exists', - 'SOA Record', - 'Primary NS', - 'Admin Email', - 'SPF Exists', - 'SPF Record', - 'DKIM Exists', - 'DKIM Records', - 'DMARC Exists', - 'DMARC Record', - 'Discovered Subdomains', - 'CNAME Records', - 'Has Wildcard DNS', - 'Hosting Provider', - 'HTTP Accessible', - 'Redirects to HTTPS', - 'Final URL', - 'Redirect Chain', - 'HTTP Error', - 'SRI Enabled', - 'Total External Resources', - 'Resources With SRI', - 'SRI Coverage %', - 'Missing SRI Count', - 'SRI Algorithms Used', - 'SRI Error' - ]) - - # Write results - for r in results: - writer.writerow([ - r['domain'], - r['timestamp'], - r['soa']['parent_domain'], - r['soa']['exists'], - r['soa'].get('record'), - r['soa'].get('primary_ns'), - r['soa'].get('admin_email'), - r['spf']['exists'], - r['spf'].get('record'), - r['dkim']['exists'], - ';'.join([f"{rec['selector']}:{rec['record']}" for rec in r['dkim']['records']]) if r['dkim']['records'] else '', - r['dmarc']['exists'], - r['dmarc'].get('record'), - ','.join(r['subdomains']['subdomains']), - ','.join([f"{k}:{v}" for k, v in r['subdomains']['cname_records'].items()]), - r['subdomains']['has_wildcard_dns'], - r['subdomains']['hosting_provider'], - r['http_redirect']['http_accessible'], - r['http_redirect']['redirects_to_https'], - r['http_redirect']['final_url'], - ' -> '.join(r['http_redirect'].get('redirect_chain', [])), - r['http_redirect']['error'], - r['sri']['sri_enabled'], - r['sri']['total_external_resources'], - r['sri']['resources_with_sri'], - r['sri']['sri_coverage_percentage'], - r['sri']['missing_sri_count'], - ','.join(r['sri']['sri_algorithms_used']) if r['sri']['sri_algorithms_used'] else '', - r['sri']['error'] - ]) + write_results_csv(results, output_file) # Optionally write filtered subdomains to a separate CSV if filtered_subdomains_file: diff --git a/domain_security_analyzer/web/__init__.py b/domain_security_analyzer/web/__init__.py new file mode 100644 index 0000000..32b9ffe --- /dev/null +++ b/domain_security_analyzer/web/__init__.py @@ -0,0 +1,17 @@ +"""Optional local web UI for Domain Security Analyzer. + +Install with the ``web`` extra:: + + pip install domain-security-analyzer[web] + +then launch:: + + domain-analyzer-web + +This subpackage imports Flask, which is only present when the ``web`` extra is +installed; importing it without Flask raises a clear ImportError. +""" + +from .app import create_app + +__all__ = ["create_app"] diff --git a/domain_security_analyzer/web/__main__.py b/domain_security_analyzer/web/__main__.py new file mode 100644 index 0000000..bd087df --- /dev/null +++ b/domain_security_analyzer/web/__main__.py @@ -0,0 +1,5 @@ +"""Enable ``python -m domain_security_analyzer.web``.""" +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/domain_security_analyzer/web/app.py b/domain_security_analyzer/web/app.py new file mode 100644 index 0000000..e1aa11f --- /dev/null +++ b/domain_security_analyzer/web/app.py @@ -0,0 +1,205 @@ +"""Flask app for the local Domain Security Analyzer web UI. + +A thin presentation layer over the analysis engine: upload a list of domains, +watch progress, download the standard 29-column CSV, and diff the two most +recent runs. Single-user / localhost by design — analysis runs in a background +thread tracked in an in-memory registry. +""" +from __future__ import annotations + +import tempfile +import threading +import uuid +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Dict, Optional + +from flask import ( + Flask, + abort, + jsonify, + redirect, + render_template, + request, + send_file, + url_for, +) + +from ..analyzer import analyze_domains_from_file +from . import runs as runs_mod + + +@dataclass +class Job: + id: str + total: int + completed: int = 0 + status: str = "running" # running | done | error + run_path: Optional[Path] = None + error: Optional[str] = None + started_at: datetime = field(default_factory=datetime.now) + + +class JobRegistry: + """In-memory registry of analysis jobs (single-user local tool).""" + + def __init__(self) -> None: + self._jobs: Dict[str, Job] = {} + self._lock = threading.Lock() + + def create(self, total: int) -> Job: + job = Job(id=uuid.uuid4().hex[:12], total=total) + with self._lock: + self._jobs[job.id] = job + return job + + def get(self, job_id: str) -> Optional[Job]: + with self._lock: + return self._jobs.get(job_id) + + def update(self, job_id: str, **fields) -> None: + with self._lock: + job = self._jobs.get(job_id) + if job: + for key, value in fields.items(): + setattr(job, key, value) + + +def parse_domains(text: str) -> list: + """Extract domains from pasted/uploaded text: one per line, '#' comments skipped.""" + domains = [] + seen = set() + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + if line.lower() not in seen: + seen.add(line.lower()) + domains.append(line) + return domains + + +def create_app() -> Flask: + app = Flask(__name__) + registry = JobRegistry() + + def _run_job(job_id: str, domains: list, max_workers: int) -> None: + """Background worker: write input, analyze, persist the run CSV.""" + try: + run_path = runs_mod.new_run_path() + with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as tmp: + tmp.write("\n".join(domains)) + input_path = tmp.name + try: + analyze_domains_from_file( + input_path, + str(run_path), + max_workers=max_workers, + progress_callback=lambda done, total: registry.update(job_id, completed=done), + ) + finally: + Path(input_path).unlink(missing_ok=True) + registry.update(job_id, status="done", run_path=run_path, completed=len(domains)) + except Exception as exc: # surface failure to the UI instead of dying silently + registry.update(job_id, status="error", error=str(exc)) + + @app.route("/") + def index(): + return render_template("index.html", runs=runs_mod.list_runs(), label=runs_mod.run_label) + + @app.route("/run", methods=["POST"]) + def run(): + text = request.form.get("domains", "") + upload = request.files.get("file") + if upload and upload.filename: + text += "\n" + upload.read().decode("utf-8", errors="replace") + + domains = parse_domains(text) + if not domains: + return render_template( + "index.html", + runs=runs_mod.list_runs(), + label=runs_mod.run_label, + error="No domains found. Paste one domain per line or upload a .txt file.", + ), 400 + + try: + max_workers = max(1, min(50, int(request.form.get("max_workers", 10)))) + except (TypeError, ValueError): + max_workers = 10 + + job = registry.create(total=len(domains)) + thread = threading.Thread( + target=_run_job, args=(job.id, domains, max_workers), daemon=True + ) + thread.start() + return redirect(url_for("run_progress", job_id=job.id)) + + @app.route("/run/") + def run_progress(job_id: str): + job = registry.get(job_id) + if not job: + abort(404) + if job.status == "done" and job.run_path is not None: + return redirect(url_for("results", run_name=job.run_path.name)) + return render_template("progress.html", job=job) + + @app.route("/run//status") + def run_status(job_id: str): + job = registry.get(job_id) + if not job: + abort(404) + payload = { + "status": job.status, + "completed": job.completed, + "total": job.total, + "error": job.error, + } + if job.status == "done" and job.run_path is not None: + payload["result_url"] = url_for("results", run_name=job.run_path.name) + return jsonify(payload) + + @app.route("/results/") + def results(run_name: str): + path = _safe_run_path(run_name) + rows = runs_mod.load_run(path) + columns = list(next(iter(rows.values())).keys()) if rows else [] + return render_template( + "results.html", + run_name=run_name, + label=runs_mod.run_label(path), + columns=columns, + rows=list(rows.values()), + ) + + @app.route("/download/") + def download(run_name: str): + path = _safe_run_path(run_name) + return send_file(path, as_attachment=True, download_name=run_name, mimetype="text/csv") + + @app.route("/changes") + def changes(): + run_files = runs_mod.list_runs() + if len(run_files) < 2: + return render_template("changes.html", insufficient=True, run_count=len(run_files)) + new_path, old_path = run_files[0], run_files[1] + diff = runs_mod.diff_runs(runs_mod.load_run(old_path), runs_mod.load_run(new_path)) + return render_template( + "changes.html", + insufficient=False, + diff=diff, + old_label=runs_mod.run_label(old_path), + new_label=runs_mod.run_label(new_path), + ) + + def _safe_run_path(run_name: str) -> Path: + """Resolve a run filename to a path inside the data dir (no traversal).""" + if "/" in run_name or "\\" in run_name or not run_name.startswith("run-"): + abort(404) + path = runs_mod.data_dir() / run_name + if not path.is_file(): + abort(404) + return path + + return app diff --git a/domain_security_analyzer/web/cli.py b/domain_security_analyzer/web/cli.py new file mode 100644 index 0000000..c4fe5d5 --- /dev/null +++ b/domain_security_analyzer/web/cli.py @@ -0,0 +1,50 @@ +"""Command-line launcher for the local web UI (``domain-analyzer-web``).""" +from __future__ import annotations + +import argparse +import sys +import threading +import webbrowser + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="domain-analyzer-web", + description="Launch the local Domain Security Analyzer web UI.", + ) + parser.add_argument("--host", default="127.0.0.1", help="Host to bind (default: 127.0.0.1)") + parser.add_argument("--port", type=int, default=8000, help="Port to bind (default: 8000)") + parser.add_argument("--open", action="store_true", help="Open the UI in a browser on start") + parser.add_argument("--debug", action="store_true", help="Run Flask in debug mode") + return parser + + +def main(argv=None) -> int: + args = build_parser().parse_args(argv) + + try: + from .app import create_app + except ModuleNotFoundError as exc: # Flask not installed + if exc.name in {"flask", "werkzeug", "jinja2"}: + print( + "The web UI requires the optional 'web' extra. Install it with:\n" + " pip install domain-security-analyzer[web]", + file=sys.stderr, + ) + return 1 + raise + + app = create_app() + url = f"http://{args.host}:{args.port}/" + print(f"Domain Security Analyzer web UI running at {url} (Ctrl+C to stop)") + + if args.open: + threading.Timer(1.0, lambda: webbrowser.open(url)).start() + + # use_reloader=False so the background browser-open and threads behave under launch. + app.run(host=args.host, port=args.port, debug=args.debug, use_reloader=False) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/domain_security_analyzer/web/runs.py b/domain_security_analyzer/web/runs.py new file mode 100644 index 0000000..737a603 --- /dev/null +++ b/domain_security_analyzer/web/runs.py @@ -0,0 +1,143 @@ +"""Run storage and diff logic for the local web UI. + +Each analysis run is persisted as a timestamped copy of the standard report +CSV under a local data directory. The "Changes" view compares the two most +recent runs and classifies per-domain field deltas into security regressions, +improvements, and informational changes (MVP: last two runs only, no database). +""" +from __future__ import annotations + +import csv +import os +from datetime import datetime +from pathlib import Path +from typing import Dict, List, Optional + +# Timestamp format used for run filenames (sortable, filesystem-safe). +RUN_TS_FORMAT = "%Y%m%d-%H%M%S" + +# Fields whose change carries a security meaning, and which direction is "good". +# For booleans, True is the healthy state; for SRI Coverage %, higher is better. +BOOLEAN_GOOD_TRUE = [ + "SOA Exists", + "SPF Exists", + "DKIM Exists", + "DMARC Exists", + "Redirects to HTTPS", + "SRI Enabled", +] +NUMERIC_HIGHER_BETTER = ["SRI Coverage %"] + +# Columns that are pure metadata / noise for a posture diff. +IGNORED_COLUMNS = {"Timestamp"} + + +def data_dir() -> Path: + """Directory where run CSVs are stored (override with DSA_DATA_DIR).""" + override = os.environ.get("DSA_DATA_DIR") + base = Path(override) if override else Path.home() / ".domain-security-analyzer" + runs = base / "runs" + runs.mkdir(parents=True, exist_ok=True) + return runs + + +def new_run_path(timestamp: Optional[datetime] = None) -> Path: + """Path for a new run CSV stamped with the given (or current) time.""" + ts = (timestamp or datetime.now()).strftime(RUN_TS_FORMAT) + return data_dir() / f"run-{ts}.csv" + + +def list_runs() -> List[Path]: + """All saved run CSVs, newest first.""" + return sorted(data_dir().glob("run-*.csv"), reverse=True) + + +def run_label(path: Path) -> str: + """Human-readable label for a run file derived from its timestamp.""" + stem = path.stem.replace("run-", "", 1) + try: + return datetime.strptime(stem, RUN_TS_FORMAT).strftime("%Y-%m-%d %H:%M:%S") + except ValueError: + return stem + + +def load_run(path: Path) -> Dict[str, Dict[str, str]]: + """Load a run CSV into a mapping of domain -> {column: value}.""" + rows: Dict[str, Dict[str, str]] = {} + with open(path, newline="") as f: + for row in csv.DictReader(f): + domain = (row.get("Domain") or "").strip() + if domain: + rows[domain] = row + return rows + + +def _classify_change(column: str, old: str, new: str) -> str: + """Return 'regression', 'improvement', or 'other' for a single field delta.""" + if column in BOOLEAN_GOOD_TRUE: + # Healthy = "True"; losing it is a regression, gaining it an improvement. + if old == "True" and new != "True": + return "regression" + if old != "True" and new == "True": + return "improvement" + return "other" + if column in NUMERIC_HIGHER_BETTER: + try: + o, n = float(old or 0), float(new or 0) + except ValueError: + return "other" + if n < o: + return "regression" + if n > o: + return "improvement" + return "other" + return "other" + + +def diff_runs(old: Dict[str, Dict[str, str]], new: Dict[str, Dict[str, str]]) -> Dict[str, object]: + """Diff two loaded runs (old -> new), classifying per-domain field changes.""" + old_domains, new_domains = set(old), set(new) + + added = sorted(new_domains - old_domains) + removed = sorted(old_domains - new_domains) + + changed: List[Dict[str, object]] = [] + for domain in sorted(old_domains & new_domains): + old_row, new_row = old[domain], new[domain] + regressions: List[Dict[str, str]] = [] + improvements: List[Dict[str, str]] = [] + other: List[Dict[str, str]] = [] + + for column in new_row: + if column in IGNORED_COLUMNS or column == "Domain": + continue + old_val = (old_row.get(column) or "").strip() + new_val = (new_row.get(column) or "").strip() + if old_val == new_val: + continue + entry = {"field": column, "old": old_val, "new": new_val} + kind = _classify_change(column, old_val, new_val) + if kind == "regression": + regressions.append(entry) + elif kind == "improvement": + improvements.append(entry) + else: + other.append(entry) + + if regressions or improvements or other: + changed.append( + { + "domain": domain, + "regressions": regressions, + "improvements": improvements, + "other": other, + } + ) + + return { + "added": added, + "removed": removed, + "changed": changed, + "regression_count": sum(len(c["regressions"]) for c in changed), + "improvement_count": sum(len(c["improvements"]) for c in changed), + } diff --git a/domain_security_analyzer/web/static/style.css b/domain_security_analyzer/web/static/style.css new file mode 100644 index 0000000..e5c6b39 --- /dev/null +++ b/domain_security_analyzer/web/static/style.css @@ -0,0 +1,104 @@ +:root { + --bg: #0f1420; + --panel: #1a2232; + --ink: #e8edf6; + --muted: #9aa7bd; + --accent: #4f8cff; + --regression: #ff6b6b; + --improvement: #46d39a; + --border: #2a3650; +} + +* { box-sizing: border-box; } + +body { + margin: 0; + font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif; + background: var(--bg); + color: var(--ink); + line-height: 1.5; +} + +header { + display: flex; + align-items: center; + justify-content: space-between; + padding: 1rem 1.5rem; + border-bottom: 1px solid var(--border); + background: var(--panel); +} + +.brand { font-weight: 700; font-size: 1.05rem; text-decoration: none; color: var(--ink); } +nav a { color: var(--muted); text-decoration: none; margin-left: 1.25rem; } +nav a:hover { color: var(--ink); } + +main { max-width: 1100px; margin: 0 auto; padding: 1.5rem; } +footer { text-align: center; color: var(--muted); font-size: 0.85rem; padding: 2rem 0; } + +h1 { margin-top: 0; } +h2 { margin-top: 2rem; } + +.muted { color: var(--muted); } +.error { color: var(--regression); font-weight: 600; } + +.card { + background: var(--panel); + border: 1px solid var(--border); + border-radius: 10px; + padding: 1.25rem; + margin: 1rem 0; +} + +label { display: block; margin: 0.75rem 0 0.25rem; font-weight: 600; } +textarea, input[type="number"], input[type="file"] { + width: 100%; + background: #0d1220; + color: var(--ink); + border: 1px solid var(--border); + border-radius: 8px; + padding: 0.6rem; + font-family: inherit; +} +textarea { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; resize: vertical; } +.row { display: flex; align-items: center; gap: 0.75rem; } +.row label { margin: 1rem 0 0; } +.row input { width: 6rem; } + +button, .button { + display: inline-block; + margin-top: 1rem; + background: var(--accent); + color: #fff; + border: none; + border-radius: 8px; + padding: 0.6rem 1.2rem; + font-size: 1rem; + font-weight: 600; + cursor: pointer; + text-decoration: none; +} +button:hover, .button:hover { filter: brightness(1.1); } + +table { width: 100%; border-collapse: collapse; } +th, td { text-align: left; padding: 0.5rem 0.6rem; border-bottom: 1px solid var(--border); } +th { color: var(--muted); font-weight: 600; white-space: nowrap; } +.runs td.actions a { margin-right: 1rem; } +.actions a { color: var(--accent); text-decoration: none; } + +.table-scroll { overflow-x: auto; border: 1px solid var(--border); border-radius: 10px; } +.results { font-size: 0.82rem; white-space: nowrap; } + +.progress { background: #0d1220; border-radius: 999px; height: 14px; overflow: hidden; border: 1px solid var(--border); } +.bar { height: 100%; background: var(--accent); transition: width 0.4s ease; } + +.summary { display: flex; gap: 0.5rem; flex-wrap: wrap; margin: 1rem 0; } +.pill { background: var(--panel); border: 1px solid var(--border); border-radius: 999px; padding: 0.25rem 0.75rem; font-size: 0.85rem; } +.pill.regression { border-color: var(--regression); color: var(--regression); } +.pill.improvement { border-color: var(--improvement); color: var(--improvement); } + +.change h3 { margin: 0 0 0.5rem; } +.delta { font-size: 0.9rem; padding: 0.15rem 0; } +.delta code { background: #0d1220; padding: 0.05rem 0.35rem; border-radius: 4px; } +.delta.regression { color: var(--regression); } +.delta.improvement { color: var(--improvement); } +.delta.other { color: var(--muted); } diff --git a/domain_security_analyzer/web/templates/base.html b/domain_security_analyzer/web/templates/base.html new file mode 100644 index 0000000..4d2fd83 --- /dev/null +++ b/domain_security_analyzer/web/templates/base.html @@ -0,0 +1,24 @@ + + + + + + {% block title %}Domain Security Analyzer{% endblock %} + + + +
+ 🛡️ Domain Security Analyzer + +
+
+ {% block content %}{% endblock %} +
+
+ Local UI · runs stored on this machine only +
+ + diff --git a/domain_security_analyzer/web/templates/changes.html b/domain_security_analyzer/web/templates/changes.html new file mode 100644 index 0000000..2bab24c --- /dev/null +++ b/domain_security_analyzer/web/templates/changes.html @@ -0,0 +1,50 @@ +{% extends "base.html" %} +{% block content %} +

Changes between runs

+ +{% if insufficient %} +

+ Need at least two runs to compare — there {{ 'is' if run_count == 1 else 'are' }} + currently {{ run_count }}. Run another analysis, then come back. +

+{% else %} +

Comparing {{ old_label }} → {{ new_label }}

+ +
+ {{ diff.regression_count }} regression(s) + {{ diff.improvement_count }} improvement(s) + {{ diff.added|length }} added + {{ diff.removed|length }} removed +
+ +{% if diff.added %} +

Domains added

+
    {% for d in diff.added %}
  • {{ d }}
    • {% endfor %}
+{% endif %} + +{% if diff.removed %} +

Domains removed

+
    {% for d in diff.removed %}
  • {{ d }}
    • {% endfor %}
+{% endif %} + +{% if diff.changed %} +

Changed domains

+{% for c in diff.changed %} +
+

{{ c.domain }}

+ {% for r in c.regressions %} +
▼ {{ r.field }}: {{ r.old }} → {{ r.new }}
+ {% endfor %} + {% for i in c.improvements %} +
▲ {{ i.field }}: {{ i.old }} → {{ i.new }}
+ {% endfor %} + {% for o in c.other %} +
• {{ o.field }}: {{ o.old }} → {{ o.new }}
+ {% endfor %} +
+{% endfor %} +{% else %} +

No per-domain field changes between these two runs.

+{% endif %} +{% endif %} +{% endblock %} diff --git a/domain_security_analyzer/web/templates/index.html b/domain_security_analyzer/web/templates/index.html new file mode 100644 index 0000000..f760cda --- /dev/null +++ b/domain_security_analyzer/web/templates/index.html @@ -0,0 +1,41 @@ +{% extends "base.html" %} +{% block content %} +

Analyze domains

+ +{% if error %}

{{ error }}

{% endif %} + +
+ + + + + + +
+ + +
+ + +
+ +

Previous runs

+{% if runs %} + + + + {% for run in runs %} + + + + + {% endfor %} + +
Run
{{ label(run) }} + View + Download CSV +
+{% else %} +

No runs yet. Your first analysis will appear here.

+{% endif %} +{% endblock %} diff --git a/domain_security_analyzer/web/templates/progress.html b/domain_security_analyzer/web/templates/progress.html new file mode 100644 index 0000000..175f1a6 --- /dev/null +++ b/domain_security_analyzer/web/templates/progress.html @@ -0,0 +1,42 @@ +{% extends "base.html" %} +{% block content %} +

Analyzing…

+ +
+
+

Starting analysis of {{ job.total }} domain(s)…

+ +
+ + +{% endblock %} diff --git a/domain_security_analyzer/web/templates/results.html b/domain_security_analyzer/web/templates/results.html new file mode 100644 index 0000000..60144b3 --- /dev/null +++ b/domain_security_analyzer/web/templates/results.html @@ -0,0 +1,27 @@ +{% extends "base.html" %} +{% block content %} +

Run results

+

{{ label }} · {{ rows|length }} domain(s)

+ +

+ Download CSV + Compare with previous run → +

+ +{% if rows %} +
+ + + {% for col in columns %}{% endfor %} + + + {% for row in rows %} + {% for col in columns %}{% endfor %} + {% endfor %} + +
{{ col }}
{{ row[col] }}
+
+{% else %} +

This run has no rows.

+{% endif %} +{% endblock %} diff --git a/pyproject.toml b/pyproject.toml index 5a14df6..7994707 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,8 +44,12 @@ dependencies = [ ] [project.optional-dependencies] +web = [ + "flask>=3.0", +] dev = [ "pytest>=7.0", + "flask>=3.0", "build", "twine", ] @@ -58,6 +62,7 @@ Changelog = "https://github.com/CallMarcus/domain-security-analyzer/blob/main/CH [project.scripts] domain-analyzer = "domain_security_analyzer.cli:main" +domain-analyzer-web = "domain_security_analyzer.web.cli:main" [tool.setuptools] py-modules = ["domain_analyzer"] @@ -65,6 +70,10 @@ py-modules = ["domain_analyzer"] [tool.setuptools.packages.find] include = ["domain_security_analyzer*"] +# Ship the web UI's templates and static assets inside the wheel. +[tool.setuptools.package-data] +"domain_security_analyzer.web" = ["templates/*.html", "static/*.css"] + # Version is derived from git tags (single source of truth). Tag `v1.2.3` # produces version `1.2.3`; commits after a tag get a dev/local suffix. [tool.setuptools_scm] diff --git a/tests/test_web.py b/tests/test_web.py new file mode 100644 index 0000000..86a2f4a --- /dev/null +++ b/tests/test_web.py @@ -0,0 +1,154 @@ +"""Tests for the optional local web UI (network-free).""" + +import csv +import time + +import pytest + +pytest.importorskip("flask") # web extra is optional + +from domain_security_analyzer.analyzer import CSV_COLUMNS, write_results_csv +from domain_security_analyzer.web import app as web_app +from domain_security_analyzer.web import runs as runs_mod +from domain_security_analyzer.web.app import create_app, parse_domains + + +@pytest.fixture(autouse=True) +def data_dir(tmp_path, monkeypatch): + """Point run storage at a temp directory for every test.""" + monkeypatch.setenv("DSA_DATA_DIR", str(tmp_path)) + return tmp_path + + +@pytest.fixture +def client(): + app = create_app() + app.config.update(TESTING=True) + return app.test_client() + + +# --- parse_domains ----------------------------------------------------------- + +def test_parse_domains_strips_blanks_comments_and_dedupes(): + text = "example.com\n\n# a comment\nEXAMPLE.com\nexample.org\n" + assert parse_domains(text) == ["example.com", "example.org"] + + +# --- diff logic -------------------------------------------------------------- + +def _row(domain, **overrides): + row = {col: "" for col in CSV_COLUMNS} + row["Domain"] = domain + row.update(overrides) + return row + + +def test_diff_detects_regression_improvement_added_removed(): + old = { + "a.com": _row("a.com", **{"SPF Exists": "True", "SRI Coverage %": "80"}), + "gone.com": _row("gone.com"), + } + new = { + "a.com": _row("a.com", **{"SPF Exists": "False", "SRI Coverage %": "90"}), + "fresh.com": _row("fresh.com"), + } + diff = runs_mod.diff_runs(old, new) + + assert diff["added"] == ["fresh.com"] + assert diff["removed"] == ["gone.com"] + assert diff["regression_count"] == 1 # SPF dropped + assert diff["improvement_count"] == 1 # SRI coverage rose + + changed = {c["domain"]: c for c in diff["changed"]}["a.com"] + assert any(r["field"] == "SPF Exists" for r in changed["regressions"]) + assert any(i["field"] == "SRI Coverage %" for i in changed["improvements"]) + + +def test_timestamp_change_is_ignored(): + old = {"a.com": _row("a.com", Timestamp="2026-01-01")} + new = {"a.com": _row("a.com", Timestamp="2026-06-25")} + diff = runs_mod.diff_runs(old, new) + assert diff["changed"] == [] + + +# --- HTTP routes ------------------------------------------------------------- + +def test_index_loads(client): + resp = client.get("/") + assert resp.status_code == 200 + assert b"Analyze domains" in resp.data + + +def test_run_without_domains_returns_400(client): + resp = client.post("/run", data={"domains": " \n# only a comment"}) + assert resp.status_code == 400 + assert b"No domains found" in resp.data + + +def _fake_analyze(monkeypatch, rows): + """Replace the engine with a synchronous CSV writer for the web worker.""" + def fake(input_file, output_file, max_workers=10, progress_callback=None, **kw): + write_results_csv(rows, output_file) + if progress_callback: + progress_callback(len(rows), len(rows)) + monkeypatch.setattr(web_app, "analyze_domains_from_file", fake) + + +def _result(domain, **overrides): + base = { + "domain": domain, + "timestamp": "2026-06-25T00:00:00", + "soa": {"exists": True, "parent_domain": domain, "record": None, "primary_ns": None, "admin_email": None}, + "spf": {"exists": True, "record": "v=spf1 -all"}, + "dkim": {"exists": False, "records": []}, + "dmarc": {"exists": True, "record": "v=DMARC1; p=none"}, + "subdomains": {"subdomains": [], "cname_records": {}, "has_wildcard_dns": False, "hosting_provider": None}, + "http_redirect": {"http_accessible": True, "redirects_to_https": True, "final_url": "https://" + domain, "error": "", "redirect_chain": []}, + "sri": {"sri_enabled": True, "total_external_resources": 1, "resources_with_sri": 1, "sri_coverage_percentage": 100, "missing_sri_count": 0, "sri_algorithms_used": ["sha384"], "error": ""}, + } + base.update(overrides) + return base + + +def _wait_for_redirect(client, location, tries=50): + for _ in range(tries): + resp = client.get(location) + if resp.status_code == 302: + return resp + time.sleep(0.02) + return resp + + +def test_full_run_flow_produces_downloadable_result(client, monkeypatch): + _fake_analyze(monkeypatch, [_result("example.com")]) + + resp = client.post("/run", data={"domains": "example.com"}) + assert resp.status_code == 302 + progress_url = resp.headers["Location"] + + # The job finishes quickly; the progress page then redirects to results. + resp = _wait_for_redirect(client, progress_url) + assert resp.status_code == 302 + results_url = resp.headers["Location"] + + page = client.get(results_url) + assert page.status_code == 200 + assert b"example.com" in page.data + + # A run CSV was persisted and is downloadable. + runs = runs_mod.list_runs() + assert len(runs) == 1 + dl = client.get("/download/" + runs[0].name) + assert dl.status_code == 200 + assert dl.headers["Content-Type"].startswith("text/csv") + + +def test_download_rejects_path_traversal(client): + assert client.get("/download/../secret").status_code == 404 + assert client.get("/download/notarun.csv").status_code == 404 + + +def test_changes_needs_two_runs(client): + resp = client.get("/changes") + assert resp.status_code == 200 + assert b"at least two runs" in resp.data