db.py

#!/usr/bin/env python3
"""
db.py — the tool an LLM is given instead of the database.

This is the whole answer to "how do we connect an AI to SQL": you do not hand it
a connection string. You hand it a small, bounded, auditable surface.

    python db.py doctor                     # prove the safety is on, before trusting it
    python db.py init                       # build the database from schema.sql + seed
    python db.py schema                     # introspect: tables, columns, keys, FKs
    python db.py query "SELECT ..."         # READ-ONLY. Enforced by the driver.
    python db.py query "... ?" -p 42        # parameters, never string concatenation
    python db.py exec  "UPDATE ..." --write # writes need an explicit gate
    python db.py exec  "UPDATE ..." --write --dry-run   # run it, then roll back
    python db.py audit                      # every write ever made, with rowcounts

Four properties that make it safe enough to give an agent:

  1. READ-ONLY IS REAL.  `query` opens file:...?mode=ro through the URI driver.
     SQLite itself refuses the write. We also reject non-SELECT text, but that
     check is the doorbell, not the lock — a regex is never the lock.
  2. WRITES ARE GATED TWICE.  `exec` refuses without --write (the gate the model
     can see) AND refuses unless ALLOW_WRITES is on in the environment (the gate
     the model cannot reach). Talking the model into passing --write gets you
     nowhere if the environment says no.
  3. NOTHING IS CONFIGURED IN THE CODE.  Paths, limits and credentials come from
     the environment; `.env` is gitignored and `.env.example` carries only
     placeholders. Secrets are redacted before anything is printed or logged.
  4. RESULTS ARE BOUNDED.  Every read is capped (--limit, default 200) so a
     careless SELECT * cannot flood a context window and cost you the session.
"""

from __future__ import annotations

import argparse
import json
import os
import random
import re
import sqlite3
import subprocess
import sys
from datetime import date, timedelta
from pathlib import Path

HERE = Path(__file__).resolve().parent


# ------------------------------------------------------------ configuration --
# Zero dependencies on purpose: this has to run on twenty machines that cannot
# all `pip install` today. Real environment variables WIN over the .env file —
# that is what lets CI, a container or a colleague's shell override the defaults
# without editing a tracked file.

def load_env(path: Path = HERE / ".env") -> None:
    if not path.exists():
        return
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        line = line.removeprefix("export ")
        key, _, val = line.partition("=")
        os.environ.setdefault(key.strip(), val.strip().strip('"').strip("'"))


def env_bool(name: str, default: bool = False) -> bool:
    return os.environ.get(name, str(default)).strip().lower() in ("1", "true", "yes", "on")


load_env()

_db_path = os.environ.get("DB_PATH", "telecom.db")
DB = Path(_db_path) if os.path.isabs(_db_path) else HERE / _db_path
AUDIT = HERE / os.environ.get("AUDIT_LOG", "audit.log")
MAX_ROWS = int(os.environ.get("MAX_ROWS", "200"))
ALLOW_WRITES = env_bool("ALLOW_WRITES", False)

# Present so the same tool shape ports to a server database. Never printed raw.
DATABASE_URL = os.environ.get("DATABASE_URL", "")

SECRET_KEY = re.compile(r"(pass(word|wd)?|secret|token|api[_-]?key|credential)", re.I)


def redact(value: str) -> str:
    """What you are allowed to put on a screen, in a log, or in an error."""
    if not value:
        return "(unset)"
    v = re.sub(r"://([^:/@]+):([^@]+)@", r"://\1:***@", value)   # DSN password
    return v if v != value else f"{value[:2]}***  [{len(value)} chars]"

WRITE_WORDS = (
    "insert", "update", "delete", "drop", "alter", "create", "replace",
    "truncate", "attach", "detach", "pragma", "vacuum", "reindex",
)


# --------------------------------------------------------------- connections --

def connect_ro() -> sqlite3.Connection:
    """A connection SQLite will not let anyone write through."""
    if not DB.exists():
        die(f"no database at {DB} — run:  python db.py init")
    con = sqlite3.connect(f"file:{DB.as_posix()}?mode=ro", uri=True)
    con.row_factory = sqlite3.Row
    return con


def connect_rw() -> sqlite3.Connection:
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA foreign_keys = ON")
    return con


def die(msg: str, code: int = 2):
    print(f"error: {msg}", file=sys.stderr)
    sys.exit(code)


def looks_like_write(sql: str) -> bool:
    head = sql.strip().lstrip("(").split(None, 1)
    return bool(head) and head[0].lower() in WRITE_WORDS


# ---------------------------------------------------------------------- init --

FIRST = ["Мария", "Георги", "Ivan", "Elena", "Dimitar", "Nikolay", "Petya",
         "Stefan", "Radost", "Kalina", "Boris"]
LAST = ["Ivanova", "Petrov", "Dimitrov", "Koleva", "Stoyanov", "Angelova",
        "Marinov", "Tsvetkova", "Nedelchev", "Popova", "Hristov"]
CITIES = ["Sofia", "Plovdiv", "Varna", "Burgas", "Ruse", "Stara Zagora"]
MODELS = ["Pixel 9", "iPhone 16", "Galaxy S25", "Xperia 1 VI", "Nothing Phone 3",
          "Galaxy A56", "iPhone 15"]
TOPICS = ["roaming charge disputed", "no data after top-up", "SIM swap request",
          "invoice unclear", "device instalment question", "coverage at home",
          "eSIM activation failed", "number portability", "VoLTE not working"]


def cmd_init(args):
    if DB.exists():
        if not args.force:
            die(f"{DB.name} already exists — pass --force to rebuild it")
        DB.unlink()

    con = connect_rw()
    con.executescript((HERE / "schema.sql").read_text(encoding="utf-8"))

    rnd = random.Random(20260803)          # seeded: same database every time
    today = date(2026, 8, 3)

    # ---- customers -----------------------------------------------------------
    people = []
    for i in range(11):
        name = f"{FIRST[i]} {LAST[i]}"
        seg = rnd.choice(["consumer", "consumer", "consumer", "business", "enterprise"])
        joined = today - timedelta(days=rnd.randint(40, 1500))
        people.append((name, f"user{i + 1}@example.test", rnd.choice(CITIES), seg,
                       joined.isoformat()))

    # The famous one. It goes in as data and comes back out as data, because
    # every insert below is parameterised. Nothing is ever concatenated.
    people.append(("Robert'); DROP TABLE customers;--", "bobby@example.test",
                   "Sofia", "consumer", "2026-02-14"))

    con.executemany(
        "INSERT INTO customers (full_name,email,city,segment,joined_on) VALUES (?,?,?,?,?)",
        people)

    # ---- plans ---------------------------------------------------------------
    plans = [
        ("S",   "Smart S",       11.99,   5,  300, 1),
        ("M",   "Smart M",       17.99,  25, 1000, 1),
        ("L",   "Smart L",       25.99,  60, 3000, 1),
        ("XL",  "Unlimited XL",  39.99, 999, 9999, 1),
        ("LEG", "Legacy 2019",    8.99,   2,  120, 0),
    ]
    con.executemany(
        "INSERT INTO plans (code,name,monthly_fee,data_gb,minutes,is_active) VALUES (?,?,?,?,?,?)",
        plans)

    # ---- addons --------------------------------------------------------------
    addons = [("Roaming EU+", 4.99), ("Extra 10GB", 6.99),
              ("Device insurance", 3.49), ("Cloud backup 200GB", 2.99)]
    con.executemany("INSERT INTO addons (name,monthly_fee) VALUES (?,?)", addons)

    # ---- subscriptions -------------------------------------------------------
    subs = []
    n_cust = len(people)
    for i in range(16):
        cust = (i % n_cust) + 1
        plan = rnd.randint(1, 5)
        started = today - timedelta(days=rnd.randint(30, 1200))
        status = rnd.choice(["active", "active", "active", "active", "suspended", "closed"])
        ended = (started + timedelta(days=rnd.randint(200, 900))).isoformat() \
            if status == "closed" else None
        subs.append((cust, plan, f"+3598{rnd.randint(10000000, 99999999)}",
                     started.isoformat(), ended, status))
    con.executemany(
        "INSERT INTO subscriptions (customer_id,plan_id,msisdn,started_on,ended_on,status)"
        " VALUES (?,?,?,?,?,?)", subs)

    # ---- subscription_addons (the many-to-many) ------------------------------
    pairs = set()
    while len(pairs) < 22:
        pairs.add((rnd.randint(1, 16), rnd.randint(1, 4)))
    con.executemany(
        "INSERT INTO subscription_addons (subscription_id,addon_id,added_on) VALUES (?,?,?)",
        [(s, a, (today - timedelta(days=rnd.randint(10, 400))).isoformat())
         for s, a in sorted(pairs)])

    # ---- devices -------------------------------------------------------------
    con.executemany(
        "INSERT INTO devices (subscription_id,model,imei,instalments_left) VALUES (?,?,?,?)",
        [(s, rnd.choice(MODELS), f"35{rnd.randint(10 ** 12, 10 ** 13 - 1)}",
          rnd.choice([0, 0, 6, 12, 18, 24]))
         for s in range(1, 17) if rnd.random() < 0.8])

    # ---- usage_daily (the big table: 30 days per subscription) --------------
    usage = []
    for s in range(1, 17):
        heavy = rnd.random() < 0.3
        for d in range(30):
            day = today - timedelta(days=29 - d)
            weekend = day.weekday() >= 5
            base = 2200 if heavy else 550
            usage.append((s, day.isoformat(),
                          max(0, int(rnd.gauss(base * (1.35 if weekend else 1.0), base * .35))),
                          max(0, int(rnd.gauss(18, 12))),
                          max(0, int(rnd.gauss(2, 3)))))
    con.executemany(
        "INSERT INTO usage_daily (subscription_id,day,data_mb,minutes,sms) VALUES (?,?,?,?,?)",
        usage)

    # ---- invoices (three periods) -------------------------------------------
    fee = dict(con.execute("SELECT id, monthly_fee FROM plans").fetchall())
    inv = []
    for s, plan_id in con.execute("SELECT id, plan_id FROM subscriptions").fetchall():
        for period in ("2026-05", "2026-06", "2026-07"):
            amount = round(fee[plan_id] + rnd.choice([0, 0, 0, 1.2, 4.8, 12.4]), 2)
            status = rnd.choice(["paid", "paid", "paid", "paid", "open", "overdue"])
            paid = f"{period}-{rnd.randint(10, 27):02d}" if status == "paid" else None
            inv.append((s, period, amount, status, paid))
    con.executemany(
        "INSERT INTO invoices (subscription_id,period,amount,status,paid_on) VALUES (?,?,?,?,?)",
        inv)

    # ---- tickets -------------------------------------------------------------
    con.executemany(
        "INSERT INTO tickets (customer_id,opened_at,topic,severity,resolved_at) VALUES (?,?,?,?,?)",
        [((i % n_cust) + 1,
          (today - timedelta(days=rnd.randint(1, 120))).isoformat(),
          rnd.choice(TOPICS), rnd.randint(1, 4),
          None if rnd.random() < 0.25 else (today - timedelta(days=rnd.randint(0, 20))).isoformat())
         for i in range(18)])

    con.commit()
    counts = {t: con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
              for t in table_names(con)}
    con.close()
    print(f"built {DB.name}")
    for t, n in counts.items():
        print(f"  {t:<22} {n:>5} rows")


# ------------------------------------------------------------- introspection --

def table_names(con) -> list[str]:
    return [r[0] for r in con.execute(
        "SELECT name FROM sqlite_master WHERE type='table' "
        "AND name NOT LIKE 'sqlite_%' ORDER BY name")]


def describe(con) -> dict:
    """Everything an LLM needs to write a correct query, and nothing more."""
    out = {"tables": [], "views": [], "foreign_keys": []}

    for t in table_names(con):
        cols = []
        for c in con.execute(f"PRAGMA table_info('{t}')"):
            cols.append({"name": c["name"], "type": c["type"] or "ANY",
                         "notnull": bool(c["notnull"]), "pk": bool(c["pk"]),
                         "default": c["dflt_value"]})
        uniques = []
        for idx in con.execute(f"PRAGMA index_list('{t}')"):
            info = [r["name"] for r in con.execute(f"PRAGMA index_info('{idx['name']}')")]
            uniques.append({"name": idx["name"], "unique": bool(idx["unique"]),
                            "columns": info, "origin": idx["origin"]})
        fks = []
        for fk in con.execute(f"PRAGMA foreign_key_list('{t}')"):
            edge = {"from_table": t, "from_column": fk["from"],
                    "to_table": fk["table"], "to_column": fk["to"] or "id",
                    "on_delete": fk["on_delete"]}
            fks.append(edge)
            out["foreign_keys"].append(edge)
        ddl = con.execute(
            "SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (t,)).fetchone()[0]
        out["tables"].append({
            "name": t, "columns": cols, "indexes": uniques, "foreign_keys": fks,
            "rows": con.execute(f"SELECT COUNT(*) FROM '{t}'").fetchone()[0],
            "ddl": ddl,
        })

    for v, sql in con.execute(
            "SELECT name, sql FROM sqlite_master WHERE type='view' ORDER BY name"):
        cols = [c["name"] for c in con.execute(f"PRAGMA table_info('{v}')")]
        out["views"].append({"name": v, "columns": cols, "ddl": sql})

    return out


def cmd_schema(args):
    con = connect_ro()
    d = describe(con)
    if args.json:
        print(json.dumps(d, indent=2, ensure_ascii=False))
        return
    for t in d["tables"]:
        print(f"\n{t['name']}  ({t['rows']} rows)")
        for c in t["columns"]:
            flags = []
            if c["pk"]:
                flags.append("PK")
            if c["notnull"]:
                flags.append("NOT NULL")
            fk = next((f for f in t["foreign_keys"] if f["from_column"] == c["name"]), None)
            if fk:
                flags.append(f"FK -> {fk['to_table']}.{fk['to_column']}")
            print(f"    {c['name']:<18} {c['type']:<10} {' · '.join(flags)}")
    print(f"\nviews: {', '.join(v['name'] for v in d['views']) or '—'}")
    print(f"foreign keys: {len(d['foreign_keys'])}")


# ---------------------------------------------------------------- read/write --

def render(rows, cols, as_json=False):
    if as_json:
        print(json.dumps([dict(zip(cols, r)) for r in rows], indent=2,
                         ensure_ascii=False, default=str))
        return
    if not rows:
        print("(0 rows)")
        return
    w = [max(len(str(c)), *(len(str(r[i])) for r in rows)) for i, c in enumerate(cols)]
    w = [min(x, 42) for x in w]
    print("  ".join(str(c)[:w[i]].ljust(w[i]) for i, c in enumerate(cols)))
    print("  ".join("-" * x for x in w))
    for r in rows:
        print("  ".join(str(r[i])[:w[i]].ljust(w[i]) for i in range(len(cols))))
    print(f"({len(rows)} rows)")


def cmd_query(args):
    sql = args.sql.strip().rstrip(";")
    if looks_like_write(sql):
        die("query is read-only - use `exec --write` for statements that change data")
    con = connect_ro()                      # the actual guarantee lives here
    try:
        cur = con.execute(f"SELECT * FROM ({sql}) LIMIT {int(args.limit)}", tuple(args.param))
    except sqlite3.Error:
        cur = con.execute(sql, tuple(args.param))   # non-wrappable (e.g. EXPLAIN)
    rows = cur.fetchmany(int(args.limit))
    cols = [d[0] for d in cur.description] if cur.description else []
    render(rows, cols, args.json)
    if len(rows) == int(args.limit):
        print(f"note: capped at {args.limit} rows - narrow the query or raise --limit")


def cmd_exec(args):
    # Gate one: the flag. An agent can pass it, and it is meant to — the flag
    # exists so that writing is never something that happens by accident.
    if not args.write:
        die("refusing to run a write without --write\n"
            f"       would run: {args.sql}")
    # Gate two: the environment. An agent cannot reach this one. It is set by
    # whoever deployed the tool, which is the whole point — no amount of clever
    # prompting changes a variable in someone else's shell.
    if not ALLOW_WRITES:
        die("ALLOW_WRITES is not enabled in this environment.\n"
            "       --write got you past the flag; the environment still says no.\n"
            "       Set ALLOW_WRITES=true in .env only if this database is meant to be writable.")
    con = connect_rw()
    try:
        cur = con.execute(args.sql, tuple(args.param))
    except sqlite3.Error as e:
        con.rollback()
        die(f"{type(e).__name__}: {e}")
    n = cur.rowcount
    if args.dry_run:
        con.rollback()
        print(f"DRY RUN - {n} row(s) would change. Rolled back, nothing written.")
        con.close()
        return
    con.commit()
    con.close()
    stamp = date.today().isoformat()
    with AUDIT.open("a", encoding="utf-8") as f:
        f.write(json.dumps({"date": stamp, "sql": args.sql,
                            "params": args.param, "rows": n}, ensure_ascii=False) + "\n")
    print(f"OK — {n} row(s) changed. Logged to {AUDIT.name}.")


# -------------------------------------------------------------------- doctor --
# A safety feature nobody has watched fail is decoration. This command tries to
# break its own guarantees and reports what happened.

def _git(*a) -> tuple[int, str]:
    try:
        p = subprocess.run(["git", *a], cwd=HERE, capture_output=True, text=True)
        return p.returncode, (p.stdout + p.stderr).strip()
    except FileNotFoundError:
        return 127, "git not installed"


def cmd_doctor(args):
    checks: list[tuple[str, bool | None, str]] = []   # name, ok (None = info), detail

    # --- secrets hygiene ------------------------------------------------------
    env_file = HERE / ".env"
    gitignore = HERE / ".gitignore"
    ignored = gitignore.exists() and any(
        l.strip() in (".env", "*.env", "/.env")
        for l in gitignore.read_text(encoding="utf-8").splitlines())
    checks.append((".env is listed in .gitignore", ignored,
                   "add a line `.env` to .gitignore" if not ignored else "yes"))

    # The check that actually matters: gitignore does nothing for a file that is
    # ALREADY tracked. This is how secrets get committed by people who did add
    # the gitignore line — just afterwards.
    rc, _ = _git("ls-files", "--error-unmatch", ".env")
    tracked = rc == 0
    checks.append((".env is NOT tracked by git", not tracked,
                   "TRACKED — run: git rm --cached .env  (and rotate anything in it)"
                   if tracked else "not tracked"))

    checks.append((".env.example exists and has no real values",
                   (HERE / ".env.example").exists(), "the file colleagues copy from"))

    # Anything secret-shaped sitting in a tracked file.
    rc, out = _git("ls-files")
    leaked = []
    if rc == 0:
        for name in out.splitlines():
            f = HERE / name
            if not f.is_file() or f.suffix in (".db", ".png", ".jpg", ".zip"):
                continue
            try:
                text = f.read_text(encoding="utf-8", errors="ignore")
            except OSError:
                continue
            for m in re.finditer(r"^\s*(\w*(?:pass\w*|secret|token|api[_-]?key))\s*=\s*(\S+)",
                                 text, re.I | re.M):
                val = m.group(2).strip("\"'")
                if val and not val.startswith(("$", "{", "<", "your", "changeme", "REPLACE")):
                    leaked.append(f"{name}: {m.group(1)}")
    checks.append(("no secret-shaped values in tracked files", not leaked,
                   "; ".join(leaked[:3]) if leaked else "clean"))

    # --- the locks ------------------------------------------------------------
    if DB.exists():
        con = connect_ro()
        n = len(table_names(con))
        checks.append((f"database reachable at {DB.name}", True, f"{n} tables"))

        # Prove the read-only connection actually refuses a write. If this check
        # ever passes silently, the lock is not a lock.
        broke = False
        try:
            con.execute("CREATE TABLE _doctor_probe (x)")
            broke = True
        except sqlite3.Error as e:
            detail = type(e).__name__
        con.close()
        checks.append(("read-only connection refuses a write", not broke,
                       "THE LOCK IS OPEN — a read query could modify data"
                       if broke else f"refused with {detail}"))
    else:
        checks.append((f"database exists at {DB.name}", False, "run: python db.py init"))

    checks.append(("write gate (ALLOW_WRITES)", None,
                   "ENABLED - this environment can write" if ALLOW_WRITES
                   else "disabled - exec will refuse even with --write"))
    checks.append(("row cap (MAX_ROWS)", None, str(MAX_ROWS)))
    checks.append(("DATABASE_URL", None, redact(DATABASE_URL)))

    # --- report ---------------------------------------------------------------
    failed = 0
    for name, ok, detail in checks:
        mark = "  ..  " if ok is None else (" PASS " if ok else " FAIL ")
        if ok is False:
            failed += 1
        print(f"[{mark}] {name:<44} {detail}")
    print()
    if failed:
        print(f"{failed} check(s) FAILED — fix before pointing this at anything real.")
        sys.exit(1)
    print("All checks passed.")


def cmd_audit(args):
    if not AUDIT.exists():
        print("(no writes recorded)")
        return
    print(AUDIT.read_text(encoding="utf-8").rstrip())


# ---------------------------------------------------------------------- main --

def main():
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = p.add_subparsers(dest="cmd", required=True)

    s = sub.add_parser("init", help="create telecom.db from schema.sql + seeded data")
    s.add_argument("--force", action="store_true")
    s.set_defaults(fn=cmd_init)

    s = sub.add_parser("schema", help="introspect the database")
    s.add_argument("--json", action="store_true")
    s.set_defaults(fn=cmd_schema)

    s = sub.add_parser("query", help="run a read-only SELECT")
    s.add_argument("sql")
    s.add_argument("-p", "--param", action="append", default=[])
    s.add_argument("--limit", default=MAX_ROWS)
    s.add_argument("--json", action="store_true")
    s.set_defaults(fn=cmd_query)

    s = sub.add_parser("exec", help="run a statement that changes data (gated)")
    s.add_argument("sql")
    s.add_argument("-p", "--param", action="append", default=[])
    s.add_argument("--write", action="store_true", help="required — the gate")
    s.add_argument("--dry-run", action="store_true", help="run it, then roll back")
    s.set_defaults(fn=cmd_exec)

    s = sub.add_parser("audit", help="show every write this tool has made")
    s.set_defaults(fn=cmd_audit)

    s = sub.add_parser("doctor", help="try to break the safety, and report what happened")
    s.set_defaults(fn=cmd_doctor)

    args = p.parse_args()
    args.fn(args)


if __name__ == "__main__":
    main()