Michael
bab2c9468c
Source-agnostic license issuance service. FastAPI app fronts a Postgres `licenses` table; the only currently-wired source is `manual` (operator mints via /internal/mint). Gumroad webhook adapter lands in PR 2. Key design points: - Signing reuses src/license/crypto.py via a COPY into the image (single source of truth — blobs minted server-side verify against the same embedded pubkey on the buyer's machine). - Source adapter Protocol (app/adapters/base.py) is the seam for Gumroad / Lemon Squeezy / Stripe in later PRs; Mint API speaks only SaleEvent / RefundEvent. - (source, source_order_id) UNIQUE composite gives idempotent webhook retries without double-mint. - JSONB type uses with_variant(JSON, 'sqlite') so the same models drive both Postgres prod and SQLite tests (no testcontainers dep). - Bearer-token auth on /internal/*; the IP-loopback guard was removed after the docker bridge made it fight legitimate prod traffic (nginx defense + Bearer remain). - Secrets resolved via *_FILE env vars pointing at /run/secrets/<name>, so passwords never appear in `docker inspect`. 21 unit tests (SQLite in-memory, StaticPool) plus a real-Postgres docker-compose smoke test in server/scripts/smoke.sh that builds the image, runs the alembic migration, mints a license, verifies the signature against the host dev pubkey, and checks the DB row. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
103 lines
3.0 KiB
Python
103 lines
3.0 KiB
Python
"""Mint core — signing, persistence, idempotency, revoke."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from app.adapters.base import SaleEvent
|
|
from app.mint import mint_from_sale, revoke_license
|
|
from app.models import License
|
|
from datatools_license.crypto import decode_blob, verify
|
|
|
|
|
|
def _sale(**overrides) -> SaleEvent:
|
|
base = dict(
|
|
source="manual",
|
|
source_order_id=None,
|
|
buyer_name="Jane Doe",
|
|
buyer_email="jane@example.com",
|
|
tier="core",
|
|
years=1,
|
|
promotion=None,
|
|
amount_paid=None,
|
|
currency="USD",
|
|
notes=None,
|
|
)
|
|
base.update(overrides)
|
|
return SaleEvent(**base)
|
|
|
|
|
|
def test_mint_persists_and_signs_verifiably(db_session):
|
|
row = mint_from_sale(db_session, _sale())
|
|
db_session.commit()
|
|
|
|
assert row.license_key.startswith("DT1-CORE-")
|
|
assert row.tier == "core"
|
|
assert row.source == "manual"
|
|
assert row.blob.startswith("DTLIC2:")
|
|
assert row.revoked_at is None
|
|
|
|
payload = decode_blob(row.blob)
|
|
sig = payload.pop("signature")
|
|
assert verify(payload, sig), "minted blob must verify against the dev pubkey"
|
|
assert payload["name"] == "Jane Doe"
|
|
assert payload["email"] == "jane@example.com"
|
|
assert payload["tier"] == "core"
|
|
|
|
|
|
def test_mint_idempotent_on_source_order_id(db_session):
|
|
"""A second mint with the same (source, source_order_id) returns
|
|
the existing row — webhook retries cannot double-mint."""
|
|
first = mint_from_sale(
|
|
db_session,
|
|
_sale(source="gumroad", source_order_id="GUM-1001"),
|
|
)
|
|
db_session.commit()
|
|
|
|
second = mint_from_sale(
|
|
db_session,
|
|
_sale(source="gumroad", source_order_id="GUM-1001", buyer_name="Different Name"),
|
|
)
|
|
db_session.commit()
|
|
|
|
assert first.license_key == second.license_key
|
|
assert second.name == "Jane Doe", "existing row is returned unchanged"
|
|
|
|
|
|
def test_manual_mints_never_dedup(db_session):
|
|
"""source_order_id=None means each manual mint creates a new row."""
|
|
a = mint_from_sale(db_session, _sale())
|
|
db_session.commit()
|
|
b = mint_from_sale(db_session, _sale())
|
|
db_session.commit()
|
|
assert a.license_key != b.license_key
|
|
|
|
|
|
def test_mint_records_commercial_metadata(db_session):
|
|
row = mint_from_sale(
|
|
db_session,
|
|
_sale(promotion="LAUNCH50", amount_paid=Decimal("79.00"), currency="USD"),
|
|
)
|
|
db_session.commit()
|
|
assert row.promotion == "LAUNCH50"
|
|
assert Decimal(str(row.amount_paid)) == Decimal("79.00")
|
|
assert row.currency == "USD"
|
|
|
|
|
|
def test_revoke_marks_row(db_session):
|
|
row = mint_from_sale(db_session, _sale())
|
|
db_session.commit()
|
|
|
|
revoked = revoke_license(db_session, license_key=row.license_key, reason="refund")
|
|
db_session.commit()
|
|
|
|
assert revoked is not None
|
|
assert revoked.revoked_at is not None
|
|
assert "refund" in (revoked.notes or "")
|
|
|
|
|
|
def test_revoke_unknown_returns_none(db_session):
|
|
assert revoke_license(db_session, license_key="DT1-CORE-no-such-key") is None
|