QSafe/server/server.py

#!/usr/bin/env python3
"""QSafe blind single-vault sync server. Standard library only; TLS is provided by Caddy."""
import argparse
import hashlib
import hmac
import json
import os
import pathlib
import secrets
import sqlite3
import struct
import time
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

MAX_BYTES = 32 * 1024 * 1024


@contextmanager
def database(path):
    db = sqlite3.connect(path, timeout=20)
    db.execute("PRAGMA journal_mode=WAL")
    db.execute("PRAGMA synchronous=FULL")
    db.execute("CREATE TABLE IF NOT EXISTS vault (id INTEGER PRIMARY KEY CHECK(id=1), etag TEXT NOT NULL, body BLOB NOT NULL, updated INTEGER NOT NULL)")
    db.execute("CREATE TABLE IF NOT EXISTS history (id INTEGER PRIMARY KEY, etag TEXT NOT NULL, body BLOB NOT NULL, updated INTEGER NOT NULL)")
    try:
        with db:
            yield db
    finally:
        db.close()


class Server(ThreadingHTTPServer):
    daemon_threads = True
    def __init__(self, address, db_path, token_hash):
        super().__init__(address, Handler)
        self.db_path, self.token_hash = db_path, token_hash
        with database(db_path) as db:
            if db.execute("PRAGMA quick_check").fetchone()[0] != "ok":
                raise RuntimeError("Database integrity check failed")


class Handler(BaseHTTPRequestHandler):
    server_version = "QSafe"
    sys_version = ""
    protocol_version = "HTTP/1.1"

    def setup(self):
        super().setup()
        self.connection.settimeout(30)

    def log_message(self, *_args):
        pass  # Do not log credentials, payloads, URLs, or card data.

    def reply(self, code, body=b"", content_type="application/json", etag=None):
        self.send_response(code)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Connection", "close")
        self.close_connection = True
        if etag:
            self.send_header("ETag", '"' + etag + '"')
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(body)

    def authenticated(self):
        value = self.headers.get("Authorization", "")
        if not value.startswith("Bearer ") or len(value) > 256:
            self.reply(401, b'{"error":"unauthorized"}')
            return False
        candidate = hashlib.sha256(value[7:].encode()).hexdigest()
        if not hmac.compare_digest(candidate, self.server.token_hash):
            self.reply(401, b'{"error":"unauthorized"}')
            return False
        return True

    def do_GET(self):
        if self.path == "/health":
            self.reply(200, b'{"service":"qsafe","protocol":1}')
            return
        if self.path != "/v1/vault":
            self.reply(404)
            return
        if not self.authenticated():
            return
        with database(self.server.db_path) as db:
            row = db.execute("SELECT etag,body FROM vault WHERE id=1").fetchone()
        if row is None:
            self.reply(404, b'{"error":"empty"}')
        else:
            self.reply(200, row[1], "application/octet-stream", row[0])

    def do_PUT(self):
        if self.path != "/v1/vault":
            self.reply(404)
            return
        if not self.authenticated():
            return
        try:
            length = int(self.headers.get("Content-Length", "-1"))
        except ValueError:
            length = -1
        if self.headers.get("Transfer-Encoding") or not 56 <= length <= MAX_BYTES:
            self.reply(413, b'{"error":"size"}')
            return
        body = self.rfile.read(length)
        if len(body) != length or body[:8] != b"QSAFE01!" or not 600_000 <= struct.unpack(">I", body[8:12])[0] <= 2_000_000:
            self.reply(400, b'{"error":"envelope"}')
            return
        etag = hashlib.sha256(body).hexdigest()
        now = int(time.time())
        with database(self.server.db_path) as db:
            db.execute("BEGIN IMMEDIATE")
            previous = db.execute("SELECT etag,body,updated FROM vault WHERE id=1").fetchone()
            match = self.headers.get("If-Match", "")
            if previous is None:
                valid = self.headers.get("If-None-Match") == "*" and not match
            else:
                valid = match == '"' + previous[0] + '"' and not self.headers.get("If-None-Match")
            if not valid:
                db.rollback()
                self.reply(412, b'{"error":"version_conflict"}')
                return
            if previous and previous[0] != etag:
                db.execute("INSERT INTO history(etag,body,updated) VALUES(?,?,?)", previous)
            db.execute("INSERT OR REPLACE INTO vault VALUES(1,?,?,?)", (etag, body, now))
            # Keep the latest 100 encrypted revisions, independent of client clocks.
            db.execute("DELETE FROM history WHERE id NOT IN (SELECT id FROM history ORDER BY id DESC LIMIT 100)")
            rows = db.execute("SELECT id,length(body) FROM history ORDER BY id DESC").fetchall()
            total = 0
            for version, size in rows:
                total += size
                if total > 256 * 1024 * 1024:
                    db.execute("DELETE FROM history WHERE id=?", (version,))
        self.reply(200, b'{"ok":true}', etag=etag)


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--bind", default="127.0.0.1")
    p.add_argument("--port", type=int, default=18792)
    p.add_argument("--data", required=True)
    p.add_argument("--token-hash-file", required=True)
    args = p.parse_args()
    os.umask(0o077)
    data = pathlib.Path(args.data)
    data.mkdir(parents=True, exist_ok=True)
    token_hash = pathlib.Path(args.token_hash_file).read_text().strip()
    if len(token_hash) != 64 or any(c not in "0123456789abcdef" for c in token_hash):
        raise ValueError("Invalid credential hash")
    Server((args.bind, args.port), str(data / "vault.sqlite3"), token_hash).serve_forever()


if __name__ == "__main__":
    main()