f179e4a015
- Added support for equivalent domains in user settings, allowing users to define custom domain groups and exclude specific global groups. - Introduced new migration script to add necessary database columns and a table for global equivalent domains. - Created a Python script to generate seed SQL for global domains, which can be executed to populate the database. - Updated API endpoints to handle equivalent domains, including retrieval and persistence of user-defined settings. - Refactored existing settings handler to remove stubs and implement full functionality for managing equivalent domains.
141 lines
4.5 KiB
Python
141 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Generate a D1/SQLite seed SQL file for Bitwarden/Vaultwarden global equivalent domains.
|
|
|
|
Why:
|
|
- Avoid bundling `global_domains.json` into the Worker (size + parse CPU).
|
|
- Store the dataset in D1 and let SQL generate the final JSON payload.
|
|
|
|
Default source:
|
|
- Downloads from Vaultwarden upstream (GitHub raw).
|
|
|
|
Usage:
|
|
python3 scripts/generate-global-domains-seed.py \
|
|
--output sql/global_domains_seed.sql
|
|
|
|
# Use a pinned URL (recommended for reproducible deploys)
|
|
python3 scripts/generate-global-domains-seed.py \
|
|
--url https://raw.githubusercontent.com/dani-garcia/vaultwarden/<tag-or-commit>/src/static/global_domains.json \
|
|
--output sql/global_domains_seed.sql
|
|
|
|
# Use a local file
|
|
python3 scripts/generate-global-domains-seed.py \
|
|
--input /path/to/global_domains.json \
|
|
--output sql/global_domains_seed.sql
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as _dt
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
DEFAULT_URL = (
|
|
"https://raw.githubusercontent.com/dani-garcia/vaultwarden/main/src/static/global_domains.json"
|
|
)
|
|
|
|
|
|
def _read_source(input_path: str | None, url: str | None) -> tuple[str, str]:
|
|
if input_path:
|
|
p = Path(input_path)
|
|
raw = p.read_text(encoding="utf-8")
|
|
return raw, str(p)
|
|
if not url:
|
|
url = DEFAULT_URL
|
|
with urllib.request.urlopen(url) as resp:
|
|
raw = resp.read().decode("utf-8")
|
|
return raw, url
|
|
|
|
|
|
def _sql_quote_single(s: str) -> str:
|
|
# SQLite single-quote escaping
|
|
return s.replace("'", "''")
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--input", help="Local path to global_domains.json")
|
|
ap.add_argument(
|
|
"--url",
|
|
default=None,
|
|
help=f"URL to download global_domains.json (default: {DEFAULT_URL})",
|
|
)
|
|
ap.add_argument(
|
|
"--output",
|
|
default="sql/global_domains_seed.sql",
|
|
help="Output SQL file path (default: sql/global_domains_seed.sql)",
|
|
)
|
|
ap.add_argument(
|
|
"--table",
|
|
default="global_equivalent_domains",
|
|
help="Target table name (default: global_equivalent_domains)",
|
|
)
|
|
args = ap.parse_args(argv)
|
|
|
|
raw, source = _read_source(args.input, args.url)
|
|
data = json.loads(raw)
|
|
if not isinstance(data, list):
|
|
raise SystemExit("Expected top-level JSON array")
|
|
|
|
rows: list[tuple[int, int, str]] = []
|
|
seen_types: set[int] = set()
|
|
for idx, item in enumerate(data):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
t = item.get("type")
|
|
domains = item.get("domains")
|
|
if not isinstance(t, int):
|
|
continue
|
|
if not isinstance(domains, list) or not all(isinstance(d, str) for d in domains):
|
|
continue
|
|
if t in seen_types:
|
|
raise SystemExit(f"Duplicate global domain type found: {t}")
|
|
seen_types.add(t)
|
|
|
|
domains_json = json.dumps(domains, ensure_ascii=False, separators=(",", ":"))
|
|
rows.append((t, idx, domains_json))
|
|
|
|
out_path = Path(args.output)
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
now = _dt.datetime.now(_dt.timezone.utc).isoformat()
|
|
header = (
|
|
f"-- Generated by scripts/generate-global-domains-seed.py at {now}\n"
|
|
f"-- Source: {source}\n"
|
|
f"-- Table: {args.table}\n"
|
|
"\n"
|
|
)
|
|
|
|
# Chunked multi-row INSERT keeps the SQL readable and efficient.
|
|
chunk_size = 200
|
|
stmts: list[str] = []
|
|
for i in range(0, len(rows), chunk_size):
|
|
chunk = rows[i : i + chunk_size]
|
|
values_sql = ",\n".join(
|
|
f"({t},{sort},'{_sql_quote_single(domains_json)}')" for (t, sort, domains_json) in chunk
|
|
)
|
|
stmts.append(
|
|
# D1 does not allow BEGIN/COMMIT in SQL scripts, so we avoid explicit transactions here.
|
|
# Use OR REPLACE to keep the operation idempotent without a destructive full-table DELETE.
|
|
f"INSERT OR REPLACE INTO {args.table} (type, sort_order, domains_json)\nVALUES\n{values_sql};"
|
|
)
|
|
|
|
# Optional cleanup: remove types that are no longer present upstream.
|
|
# If this statement fails, the table will still contain a superset of valid groups (harmless).
|
|
type_list = ",".join(str(t) for (t, _, _) in rows)
|
|
stmts.append(f"DELETE FROM {args.table} WHERE type NOT IN ({type_list});")
|
|
stmts.append("")
|
|
|
|
out_path.write_text(header + "\n".join(stmts), encoding="utf-8")
|
|
print(f"Wrote {len(rows)} global domain groups to {out_path}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|
|
|
|
|