git / moonbite-network / tests/test_audit_infra_2026_10_09.py · raw

"""Web-tier regression tests for the 2026-10-09 infrastructure audit fixes.  H1  invoice creation is token-gated and bound to the merchant's own address  H2  invoice GET / qr.svg are rate-limited; scantxoutset shares a global budget  L1  manage token accepted only in the X-Merchant-Token header, never the URL  L6  Permissions-Policy header on every response  M5  read-only proxy allows sendrawtransaction with a strict hex check  NEW-6  /api/wall rejects non-integer limit/offset with 400, not 500"""import importlibimport osimport sysimport pytestimport merchantsimport web_appfrom transaction import generate_keypair, pubkey_hashfrom wallet import address_from_pubkey_hash@pytest.fixture()def client():    web_app.app.config["TESTING"] = True    return web_app.app.test_client()@pytest.fixture()def fresh_merchants(tmp_path):    merchants._conn = None    merchants._DB_PATH = tmp_path / "merchants.db"    yield    if merchants._conn is not None:        merchants._conn.close()    merchants._conn = None    importlib.reload(merchants)def _demo_address() -> str:    _, pub = generate_keypair()    return address_from_pubkey_hash(pubkey_hash(pub))def _register(client, addr=None):    addr = addr or _demo_address()    r = client.post(        "/api/merchants",        json={"name": "Audit Shop", "category": "goods", "mbite_address": addr},    )    assert r.status_code == 201, r.get_json()    m = r.get_json()["merchant"]    return m["id"], m["manage_token"], addr# --------------------------------------------------------------------------- H1def test_invoice_create_requires_token(client, fresh_merchants):    mid, token, addr = _register(client)    anon = client.post("/api/merchant/invoice", json={"address": addr, "amount": "1"})    assert anon.status_code == 403    no_token = client.post(        "/api/merchant/invoice", json={"merchant_id": mid, "amount": "1"}    )    assert no_token.status_code == 403    bad = client.post(        "/api/merchant/invoice",        json={"merchant_id": mid, "amount": "1"},        headers={"X-Merchant-Token": "wrong"},    )    assert bad.status_code == 403    ok = client.post(        "/api/merchant/invoice",        json={"merchant_id": mid, "amount": "1"},        headers={"X-Merchant-Token": token},    )    assert ok.status_code == 201    assert ok.get_json()["invoice"]["address"] == addr    # Body-field token is also accepted for POST.    ok2 = client.post(        "/api/merchant/invoice",        json={"merchant_id": mid, "amount": "1", "manage_token": token},    )    assert ok2.status_code == 201def test_invoice_create_rejects_foreign_address(client, fresh_merchants):    mid, token, _ = _register(client)    other = _demo_address()    r = client.post(        "/api/merchant/invoice",        json={"merchant_id": mid, "amount": "1", "address": other},        headers={"X-Merchant-Token": token},    )    assert r.status_code == 400    assert "not registered" in r.get_json()["message"]# --------------------------------------------------------------------------- L1def test_manage_token_in_query_string_is_ignored(client, fresh_merchants):    mid, token, _ = _register(client)    r = client.get(f"/api/merchant/invoices?merchant_id={mid}&manage_token={token}")    assert r.status_code == 403    r2 = client.get(        f"/api/merchant/invoices?merchant_id={mid}", headers={"X-Merchant-Token": token}    )    assert r2.status_code == 200# --------------------------------------------------------------------------- H2def test_invoice_get_and_qr_are_rate_limited(client, fresh_merchants):    mid, token, _ = _register(client)    inv = client.post(        "/api/merchant/invoice",        json={"merchant_id": mid, "amount": "1"},        headers={"X-Merchant-Token": token},    ).get_json()["invoice"]    web_app._rl_hits.clear()    web_app._RATE_DISABLED = False    try:        codes = [            client.get(f"/api/merchant/invoice/{inv['id']}").status_code for _ in range(35)        ]        qr_codes = [            client.get(f"/api/merchant/invoice/{inv['id']}/qr.svg").status_code            for _ in range(35)        ]    finally:        web_app._RATE_DISABLED = True        web_app._rl_hits.clear()    assert codes.count(200) == 30 and codes.count(429) == 5    # QR may be 501 if the encoder is absent; it must never exceed 30 non-429s.    assert qr_codes.count(429) == 5class _FakeRpc:    def __init__(self, total_amount="1.0"):        self.total_amount = total_amount        self.calls = 0    def scantxoutset(self, action, scanobjects):        self.calls += 1        return {"success": True, "total_amount": self.total_amount}def test_scan_budget_exhausts_then_serves_stale_cache(monkeypatch):    fake = _FakeRpc("2.0")    monkeypatch.setattr(web_app, "_get_merchant_rpc", lambda: fake)    monkeypatch.setattr(web_app, "_MERCHANT_RECV_TTL", 0)  # no fresh-cache hits    monkeypatch.setattr(web_app, "_SCAN_BUCKET_CAPACITY", 3)    monkeypatch.setattr(web_app, "_SCAN_BUCKET_WINDOW", 1000.0)  # effectively no refill    monkeypatch.setattr(merchants, "UNITS_PER_COIN", 100)    web_app._merchant_recv_cache.clear()    web_app._scan_budget_reset()    # Three distinct never-seen addresses consume the whole bucket.    for a in ("addrA", "addrB", "addrC"):        assert web_app.received_at_address_rpc(a) == 200    assert fake.calls == 3    # A fourth, uncached address: budget spent -> raise, do NOT fabricate 0.    with pytest.raises(web_app.ScanBudgetExhausted) as ei:        web_app.received_at_address_rpc("addrD")    assert ei.value.retry_after >= 1    # A previously scanned address still answers from its (stale) cache entry.    assert web_app.received_at_address_rpc("addrA") == 200    assert fake.calls == 3  # no extra scans slipped through    web_app._scan_budget_reset()def test_budget_exhaustion_is_503_with_retry_after(client, fresh_merchants, monkeypatch):    mid, token, _ = _register(client)    inv = client.post(        "/api/merchant/invoice",        json={"merchant_id": mid, "amount": "1"},        headers={"X-Merchant-Token": token},    ).get_json()["invoice"]    def boom(_addr):        raise web_app.ScanBudgetExhausted(retry_after=7)    monkeypatch.setattr(web_app, "merchant_received_lookup", boom)    r = client.get(f"/api/merchant/invoice/{inv['id']}")    assert r.status_code == 503    assert r.headers["Retry-After"] == "7"    r2 = client.get(f"/api/merchant/invoice/{inv['id']}/qr.svg")    assert r2.status_code == 503def test_chain_address_route_uses_budget(client, monkeypatch):    monkeypatch.setattr(web_app, "_merchant_use_rpc", lambda: True)    monkeypatch.setattr(web_app, "_valid_receiving_address", lambda a: True)    monkeypatch.setattr(web_app, "_get_merchant_rpc", lambda: _FakeRpc())    def boom(rpc, addr):        raise web_app.ScanBudgetExhausted(retry_after=4)    monkeypatch.setattr(web_app, "_budgeted_scantxoutset", boom)    web_app._chain_addr_cache.clear()    r = client.get("/api/chain/address/moon1budgettest")    assert r.status_code == 503    assert r.headers["Retry-After"] == "4"# --------------------------------------------------------------------------- L6def test_permissions_policy_header_present(client):    r = client.get("/")    pp = r.headers.get("Permissions-Policy", "")    for feat in ("camera=()", "microphone=()", "geolocation=()", "payment=()", "usb=()"):        assert feat in pp    # WebAuthn must stay available to the wallet.    assert "publickey-credentials-get" not in pp    # HSTS stays app-level (no nginx template in deploy/ sets it).    assert "Strict-Transport-Security" in r.headers# --------------------------------------------------------------------------- NEW-6def test_wall_rejects_non_integer_paging(client):    assert client.get("/api/wall?limit=abc").status_code == 400    assert client.get("/api/wall?offset=1.5").status_code == 400    assert client.get("/api/wall?limit=5&offset=0").status_code == 200# --------------------------------------------------------------------------- M5def test_proxy_sendrawtransaction_hex_validation():    sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(__file__)), "explorer"))    import rpc_readonly_proxy as proxy    assert "sendrawtransaction" in proxy.ALLOWED_METHODS    assert "sendtoaddress" not in proxy.ALLOWED_METHODS    v = proxy._validate_params    assert v("sendrawtransaction", ["deadbeef"]) is None    assert v("sendrawtransaction", ["zz"]) is not None    assert v("sendrawtransaction", ["abc"]) is not None          # odd length    assert v("sendrawtransaction", [""]) is not None    assert v("sendrawtransaction", ["aa", 0.1]) is not None      # no maxfeerate    assert v("sendrawtransaction", {"hexstring": "aa"}) is not None    assert v("sendrawtransaction", [123]) is not None    assert v("getblockcount", []) is Nonedef test_merchant_page_has_no_fabricated_data(client):    r = client.get("/merchants")    assert r.status_code == 200    body = r.get_data(as_text=True)    assert "Math.random" not in body    assert "2%" not in body and "98%" not in body    assert "create-invoice" not in body  # retired mock endpoint not advertised    assert "/api/merchant/invoice" in body    assert "X-Merchant-Token" in body