dove-core 0.1.1

"""Unit tests for the dove access gate — the request (file-request) routes.

Runs with in-process fakes for DynamoDB and S3 (no AWS, no network). The gate
module is imported from ``gate.py`` after the environment it reads at import
time is set, then its module-level ``_ddb``/``_s3`` clients are swapped for the
fakes below.

    python3 -m unittest -v        # from assets/
    python3 test_gate.py
"""

import hashlib
import hmac
import importlib.util
import json
import os
import time
import unittest
from pathlib import Path

# The gate reads BUCKET/TABLE at import and mints/validates ids against
# GATE_SECRET; set all three before importing. A region + dummy creds keep
# boto3.client(...) from raising at import (the clients are swapped out anyway).
os.environ.setdefault("BUCKET", "dove-test-bucket")
os.environ.setdefault("TABLE", "dove-test-table")
GATE_SECRET_HEX = "aa" * 32
os.environ["GATE_SECRET"] = GATE_SECRET_HEX
os.environ.pop("GATE_SECRET_PARAM", None)
os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
os.environ.setdefault("AWS_ACCESS_KEY_ID", "testing")
os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "testing")

_SECRET = bytes.fromhex(GATE_SECRET_HEX)

_spec = importlib.util.spec_from_file_location("dove_gate", Path(__file__).parent / "gate.py")
gate = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(gate)


# --- helpers ---------------------------------------------------------------

def mint_id(nonce=b"\x01\x02\x03\x04\x05\x06\x07\x08"):
    """A well-formed id this gate's MAC accepts."""
    mac = hmac.new(_SECRET, nonce, hashlib.sha256).digest()[:8]
    return (nonce + mac).hex()


def pin_hash(rid, pin):
    return hashlib.sha256(f"{rid}:{pin}".encode()).hexdigest()


def N(v):
    return {"N": str(v)}


def S(v):
    return {"S": str(v)}


# --- fakes -----------------------------------------------------------------

class _ConditionalCheckFailedException(Exception):
    pass


class _Exceptions:
    ConditionalCheckFailedException = _ConditionalCheckFailedException


def _split_top(s):
    """Split on top-level commas (ignoring commas nested in parentheses, e.g.
    ``if_not_exists(pin_attempts, :z)``)."""
    parts, depth, cur = [], 0, ""
    for ch in s:
        if ch == "(":
            depth += 1
            cur += ch
        elif ch == ")":
            depth -= 1
            cur += ch
        elif ch == "," and depth == 0:
            parts.append(cur)
            cur = ""
        else:
            cur += ch
    if cur.strip():
        parts.append(cur)
    return [p.strip() for p in parts]


class FakeDdb:
    """A tiny DynamoDB stand-in: get_item + a conditional update_item that
    understands exactly the expressions the gate emits."""

    def __init__(self):
        self.items = {}
        self.exceptions = _Exceptions()

    def put(self, item):
        self.items[item["id"]["S"]] = dict(item)

    def get_item(self, TableName, Key):
        item = self.items.get(Key["id"]["S"])
        return {"Item": dict(item)} if item is not None else {}

    def update_item(self, TableName, Key, UpdateExpression, ConditionExpression=None,
                    ExpressionAttributeValues=None, ReturnValues=None):
        eav = ExpressionAttributeValues or {}
        item = self.items.get(Key["id"]["S"])
        cur = item if item is not None else {}
        if ConditionExpression and not self._cond(ConditionExpression, cur, eav):
            raise _ConditionalCheckFailedException("condition failed")
        if item is None:
            item = {"id": Key["id"]}
            self.items[Key["id"]["S"]] = item
        self._apply(UpdateExpression, item, eav)
        return {"Attributes": dict(item)}

    def _cond(self, expr, item, eav):
        for or_term in expr.split(" OR "):
            if all(self._atom(a.strip(), item, eav) for a in or_term.split(" AND ")):
                return True
        return False

    def _atom(self, a, item, eav):
        if a.startswith("attribute_not_exists(") and a.endswith(")"):
            return a[len("attribute_not_exists("):-1].strip() not in item
        if a.startswith("attribute_exists(") and a.endswith(")"):
            return a[len("attribute_exists("):-1].strip() in item
        for op in ("<", ">"):
            if f" {op} " in a:
                lhs, rhs = a.split(f" {op} ")
                lv = int(item.get(lhs.strip(), {"N": "0"})["N"])
                rv = int(eav[rhs.strip()]["N"])
                return lv < rv if op == "<" else lv > rv
        raise ValueError(f"unhandled condition atom: {a!r}")

    def _apply(self, expr, item, eav):
        assert expr.startswith("SET "), expr
        for assign in _split_top(expr[4:]):
            lhs, rhs = assign.split(" = ", 1)
            item[lhs.strip()] = self._rhs(rhs.strip(), item, eav)

    def _rhs(self, rhs, item, eav):
        if rhs.startswith("if_not_exists("):
            inner, _, plus = rhs.partition(") + ")
            attr, dref = [x.strip() for x in inner[len("if_not_exists("):].split(",")]
            base = int(item.get(attr, eav[dref])["N"])
            return {"N": str(base + int(eav[plus.strip()]["N"]))}
        if " - " in rhs:
            attr, ref = rhs.split(" - ")
            return {"N": str(int(item[attr.strip()]["N"]) - int(eav[ref.strip()]["N"]))}
        if rhs.startswith(":"):
            return eav[rhs]
        raise ValueError(f"unhandled update rhs: {rhs!r}")


class FakeS3:
    def __init__(self):
        self.objects = {}  # key -> size
        self.posts = []

    def head_object(self, Bucket, Key):
        if Key not in self.objects:
            raise Exception("404 Not Found")
        return {"ContentLength": self.objects[Key]}

    def generate_presigned_post(self, Bucket, Key, Fields=None, Conditions=None, ExpiresIn=None):
        self.posts.append({"Bucket": Bucket, "Key": Key, "Conditions": Conditions,
                           "ExpiresIn": ExpiresIn})
        return {"url": f"https://s3.example/{Bucket}",
                "fields": {"key": Key, "policy": "b64policy", "x-amz-signature": "sig"}}


# --- test base -------------------------------------------------------------

class GateTest(unittest.TestCase):
    def setUp(self):
        self.ddb = FakeDdb()
        self.s3 = FakeS3()
        gate._ddb = self.ddb
        gate._s3 = self.s3
        self.now = int(time.time())

    def request_row(self, rid, pin=None, uploads_remaining=1, expires_in=3600,
                    pin_attempts=0, upload_attempts=0, s3_key=None, upload_meta=None,
                    size=None, meta="trustblob"):
        item = {
            "id": S(rid),
            "kind": S("request"),
            "uploads_remaining": N(uploads_remaining),
            "uploads_total": N(1),
            "expires_at": N(self.now + expires_in),
            "meta": S(meta),
            "pin_attempts": N(pin_attempts),
            "upload_attempts": N(upload_attempts),
        }
        if pin is not None:
            item["pin_hash"] = S(pin_hash(rid, pin))
        if s3_key is not None:
            item["s3_key"] = S(s3_key)
        if upload_meta is not None:
            item["upload_meta"] = S(upload_meta)
        if size is not None:
            item["size"] = N(size)
        self.ddb.put(item)
        return item

    def share_row(self, sid, pin=None, downloads_remaining=1, expires_in=3600, pin_attempts=0):
        item = {
            "id": S(sid),
            "downloads_remaining": N(downloads_remaining),
            "downloads_total": N(1),
            "expires_at": N(self.now + expires_in),
            "size": N(1024),
            "meta": S("trustblob"),
            "s3_key": S(f"blob/{sid}"),
            "pin_attempts": N(pin_attempts),
        }
        if pin is not None:
            item["pin_hash"] = S(pin_hash(sid, pin))
        self.ddb.put(item)
        return item

    def call(self, route, rid, pin=None, body=None, method="GET"):
        event = {"rawPath": f"/{route}/{rid}", "requestContext": {"domainName": "gate.example"}}
        if pin is not None:
            event["queryStringParameters"] = {"pin": pin}
        if body is not None:
            event["body"] = body
        return gate.handler(event, None)

    def json_body(self, resp):
        return json.loads(resp["body"])


# --- /r (request upload page) -----------------------------------------------

class RequestPageTest(GateTest):
    def test_serves_request_page_with_origin_injected(self):
        resp = self.call("r", "anything")
        self.assertEqual(resp["statusCode"], 200)
        self.assertIn("text/html", resp["headers"]["content-type"])
        # A distinctive marker from request.html (not share.html).
        self.assertIn("A file was requested from you", resp["body"])
        # __OGBASE__ was replaced with this gate's own origin.
        self.assertNotIn("__OGBASE__", resp["body"])
        self.assertIn("https://gate.example/og-request.png", resp["body"])


# --- MAC gating ------------------------------------------------------------

class MacGateTest(GateTest):
    def test_forged_id_rejected_on_request_routes(self):
        forged = "de" * 16  # 32 hex chars, wrong MAC
        for route in ("rmeta", "up", "done"):
            resp = self.call(route, forged, pin="1234")
            self.assertEqual(resp["statusCode"], 403, route)

    def test_wrong_length_id_rejected(self):
        for route in ("rmeta", "up", "done"):
            resp = self.call(route, "abc123", pin="1234")
            self.assertEqual(resp["statusCode"], 403, route)


# --- /rmeta ----------------------------------------------------------------

class RmetaTest(GateTest):
    def test_missing_is_404(self):
        resp = self.call("rmeta", mint_id())
        self.assertEqual(resp["statusCode"], 404)

    def test_waiting(self):
        rid = mint_id()
        self.request_row(rid, pin="1234")
        body = self.json_body(self.call("rmeta", rid))
        self.assertEqual(body["status"], "waiting")
        self.assertIsNone(body["reason"])
        self.assertTrue(body["pin_required"])
        self.assertFalse(body["locked"])
        self.assertEqual(body["uploads_remaining"], 1)
        self.assertEqual(body["meta"], "trustblob")
        self.assertNotIn("name_meta", body)
        self.assertNotIn("size", body)

    def test_received(self):
        rid = mint_id()
        self.request_row(rid, s3_key=f"req/{rid}", upload_meta="namemeta", size=4242)
        body = self.json_body(self.call("rmeta", rid))
        self.assertEqual(body["status"], "received")
        self.assertIsNone(body["reason"])
        self.assertEqual(body["name_meta"], "namemeta")
        self.assertEqual(body["size"], 4242)

    def test_failed_locked(self):
        rid = mint_id()
        self.request_row(rid, pin="1234", pin_attempts=gate.MAX_PIN_ATTEMPTS)
        body = self.json_body(self.call("rmeta", rid))
        self.assertEqual(body["status"], "failed")
        self.assertEqual(body["reason"], "locked")
        self.assertTrue(body["locked"])

    def test_failed_rate_limited(self):
        rid = mint_id()
        self.request_row(rid, upload_attempts=gate.MAX_UPLOAD_ATTEMPTS)
        body = self.json_body(self.call("rmeta", rid))
        self.assertEqual(body["status"], "failed")
        self.assertEqual(body["reason"], "rate-limited")

    def test_failed_expired(self):
        rid = mint_id()
        self.request_row(rid, expires_in=-10)
        body = self.json_body(self.call("rmeta", rid))
        self.assertEqual(body["status"], "failed")
        self.assertEqual(body["reason"], "expired")

    def test_received_wins_over_expired(self):
        rid = mint_id()
        self.request_row(rid, expires_in=-10, s3_key=f"req/{rid}", upload_meta="nm", size=9)
        body = self.json_body(self.call("rmeta", rid))
        self.assertEqual(body["status"], "received")


# --- /up -------------------------------------------------------------------

class UpTest(GateTest):
    def test_correct_pin_hands_out_post_without_spending_budget(self):
        rid = mint_id()
        self.request_row(rid, pin="1234")
        resp = self.call("up", rid, pin="1234")
        self.assertEqual(resp["statusCode"], 200)
        body = self.json_body(resp)
        self.assertEqual(body["key"], f"req/{rid}")
        self.assertIn("url", body)
        self.assertIn("fields", body)
        # budget untouched; attempt counted; content-length-range applied
        stored = self.ddb.items[rid]
        self.assertEqual(int(stored["uploads_remaining"]["N"]), 1)
        self.assertEqual(int(stored["upload_attempts"]["N"]), 1)
        self.assertEqual(self.s3.posts[-1]["Conditions"],
                         [["content-length-range", 0, gate.MAX_UPLOAD]])

    def test_no_pin_request_hands_out_post(self):
        rid = mint_id()
        self.request_row(rid)  # no pin
        resp = self.call("up", rid)
        self.assertEqual(resp["statusCode"], 200)

    def test_absent_pin_is_401(self):
        rid = mint_id()
        self.request_row(rid, pin="1234")
        resp = self.call("up", rid)
        self.assertEqual(resp["statusCode"], 401)

    def test_wrong_pin_is_401(self):
        rid = mint_id()
        self.request_row(rid, pin="1234")
        resp = self.call("up", rid, pin="9999")
        self.assertEqual(resp["statusCode"], 401)
        self.assertEqual(int(self.ddb.items[rid]["pin_attempts"]["N"]), 1)

    def test_at_ceiling_is_429(self):
        rid = mint_id()
        self.request_row(rid, pin="1234", upload_attempts=gate.MAX_UPLOAD_ATTEMPTS)
        resp = self.call("up", rid, pin="1234")
        self.assertEqual(resp["statusCode"], 429)
        self.assertEqual(self.json_body(resp)["error"], "rate-limited")

    def test_already_received_is_410(self):
        rid = mint_id()
        self.request_row(rid, s3_key=f"req/{rid}")
        resp = self.call("up", rid)
        self.assertEqual(resp["statusCode"], 410)

    def test_expired_is_410(self):
        rid = mint_id()
        self.request_row(rid, expires_in=-10)
        resp = self.call("up", rid)
        self.assertEqual(resp["statusCode"], 410)

    def test_missing_is_404(self):
        resp = self.call("up", mint_id())
        self.assertEqual(resp["statusCode"], 404)

    def test_share_row_on_up_is_404(self):
        sid = mint_id()
        self.share_row(sid)  # kind absent
        resp = self.call("up", sid)
        self.assertEqual(resp["statusCode"], 404)


# --- /done -----------------------------------------------------------------

class DoneTest(GateTest):
    def test_no_object_is_410(self):
        rid = mint_id()
        self.request_row(rid)
        resp = self.call("done", rid, body=json.dumps({"upload_meta": "nm"}), method="POST")
        self.assertEqual(resp["statusCode"], 410)
        self.assertEqual(self.json_body(resp)["error"], "no upload found")

    def test_finalize_sets_fields_and_decrements(self):
        rid = mint_id()
        self.request_row(rid, pin="1234")
        self.s3.objects[f"req/{rid}"] = 5000
        resp = self.call("done", rid, pin="1234",
                         body=json.dumps({"upload_meta": "namemeta"}), method="POST")
        self.assertEqual(resp["statusCode"], 200)
        self.assertTrue(self.json_body(resp)["ok"])
        stored = self.ddb.items[rid]
        self.assertEqual(stored["s3_key"]["S"], f"req/{rid}")
        self.assertEqual(int(stored["size"]["N"]), 5000)
        self.assertEqual(stored["upload_meta"]["S"], "namemeta")
        self.assertIn("received_at", stored)
        self.assertEqual(int(stored["uploads_remaining"]["N"]), 0)

    def test_idempotent_second_call(self):
        rid = mint_id()
        self.request_row(rid)
        self.s3.objects[f"req/{rid}"] = 5000
        first = self.call("done", rid, body=json.dumps({"upload_meta": "nm"}), method="POST")
        self.assertEqual(first["statusCode"], 200)
        # a second finalize must not decrement again nor overwrite
        second = self.call("done", rid, body=json.dumps({"upload_meta": "OTHER"}), method="POST")
        self.assertEqual(second["statusCode"], 200)
        self.assertTrue(self.json_body(second)["ok"])
        stored = self.ddb.items[rid]
        self.assertEqual(int(stored["uploads_remaining"]["N"]), 0)
        self.assertEqual(stored["upload_meta"]["S"], "nm")

    def test_missing_row_is_404(self):
        resp = self.call("done", mint_id(), body=json.dumps({"upload_meta": "nm"}), method="POST")
        self.assertEqual(resp["statusCode"], 404)


# --- /verify regression (share) + kind-aware (request) ---------------------

class VerifyTest(GateTest):
    def test_share_correct_pin_still_ok(self):
        sid = mint_id()
        self.share_row(sid, pin="1234")
        resp = self.call("verify", sid, pin="1234")
        self.assertEqual(resp["statusCode"], 200)
        self.assertTrue(self.json_body(resp)["ok"])

    def test_share_exhausted_is_gone(self):
        sid = mint_id()
        self.share_row(sid, pin="1234", downloads_remaining=0)
        resp = self.call("verify", sid, pin="1234")
        self.assertEqual(resp["statusCode"], 410)

    def test_share_no_pin_ok(self):
        sid = mint_id()
        self.share_row(sid)
        resp = self.call("verify", sid)
        self.assertEqual(resp["statusCode"], 200)

    def test_request_correct_pin_ok(self):
        rid = mint_id()
        self.request_row(rid, pin="1234")
        resp = self.call("verify", rid, pin="1234")
        self.assertEqual(resp["statusCode"], 200)
        self.assertTrue(self.json_body(resp)["ok"])

    def test_request_already_received_is_gone(self):
        rid = mint_id()
        self.request_row(rid, pin="1234", s3_key=f"req/{rid}")
        resp = self.call("verify", rid, pin="1234")
        self.assertEqual(resp["statusCode"], 410)

    def test_request_wrong_pin_is_401(self):
        rid = mint_id()
        self.request_row(rid, pin="1234")
        resp = self.call("verify", rid, pin="0000")
        self.assertEqual(resp["statusCode"], 401)


if __name__ == "__main__":
    unittest.main(verbosity=2)