2ro
38d7d0b387
check_giftwrap_recipient accepted a p tag of len 64 that parsed as hex in
any case, but the recipient-only READ gate compares #p to the authed
pubkey as a case-sensitive lowercase string. An uppercase/mixed-case p
tag therefore admitted fine but was permanently undeliverable, a payment
black hole. Now the p value must match ^[0-9a-f]{64}$.
A new check_giftwrap_tags guard (1059 only) rejects any tag other than
the single p tag, since a well-formed NIP-59 gift wrap legitimately
carries just that one tag. Closes the anti-bloat gap where a valid p tag
plus thousands of junk tags (up to the 128 KiB event ceiling) passed
admission unchecked.
Both checks are kind-1059-only and sit alongside the existing gift-wrap
retention/shape checks in CHECKS; no other kind's handling changes.
468 lines
19 KiB
Python
468 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for the floonet write policy.
|
|
|
|
Run from the plugin directory: python3 test_policy.py
|
|
|
|
Two layers:
|
|
* unit tests over decide()/the check functions (whitelist, auth, paid,
|
|
fail-closed behavior), with the paid lookup stubbed by a real local HTTP
|
|
server standing in for the name authority;
|
|
* a subprocess pipe test that runs the plugin exactly the way strfry does
|
|
(JSONL on stdin, JSONL on stdout) and asserts accept/reject decisions.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
import floonet_writepolicy as wp
|
|
|
|
PLUGIN = os.path.join(os.path.dirname(os.path.abspath(__file__)), "floonet_writepolicy.py")
|
|
PK = "a" * 64
|
|
DEFAULT_KINDS = (
|
|
0, 1, 3, 5, 7, 13, 14, 16, 17, 1059, 1111, 10000, 10002, 10050, 24133,
|
|
27235, 30000, 30003, 30023, 30078, 30402, 30405, 30406, 31990,
|
|
)
|
|
# npub for PK (the 32-byte key 0xaa..aa); another key for "unauthorized".
|
|
PK_NPUB = "npub1424242424242424242424242424242424242424242424242424qamrcaj"
|
|
OTHER_PK = "b" * 64
|
|
|
|
|
|
def req(kind, authed=None, event_id="e1", pubkey=PK, tags=None):
|
|
"""A request shaped exactly like strfry's plugin input. Kind 1059 (gift
|
|
wrap) defaults to a single well-formed `p` tag so existing tests that
|
|
only care about other checks keep passing the gift-wrap shape guard;
|
|
pass tags=[...] explicitly to exercise that guard itself."""
|
|
if tags is None:
|
|
tags = [["p", OTHER_PK]] if kind == 1059 else []
|
|
r = {
|
|
"type": "new",
|
|
"event": {"id": event_id, "pubkey": pubkey, "kind": kind, "tags": tags, "content": ""},
|
|
"receivedAt": 1700000000,
|
|
"sourceType": "IP4",
|
|
"sourceInfo": "203.0.113.7",
|
|
}
|
|
if authed is not None:
|
|
r["authed"] = authed
|
|
return r
|
|
|
|
|
|
def cfg(**over):
|
|
base = wp.load_config(env={})
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
class KindWhitelist(unittest.TestCase):
|
|
def test_default_allowed_kinds_accepted(self):
|
|
# Authorize PK so the locked public-note kinds (1, 30023) also pass;
|
|
# this test only exercises the kind whitelist.
|
|
c = cfg(authorized_authors={PK})
|
|
for kind in DEFAULT_KINDS:
|
|
reply = wp.decide(req(kind), c)
|
|
self.assertEqual(reply["action"], "accept", "kind %d" % kind)
|
|
self.assertEqual(reply["id"], "e1")
|
|
|
|
def test_disallowed_kinds_rejected(self):
|
|
# 25910 (ContextVM) rides inside 1059 gift wraps only;
|
|
# 30017/30018 (legacy NIP-15) come from sellers' own relays;
|
|
# 9735 (zap) is dead in the GRIN-only fork. All stay rejected.
|
|
# (30023 is now whitelisted but author-locked; see PublicNoteLock.)
|
|
for kind in (4, 6, 9735, 1058, 1060, 25910, 30017, 30018, 22242, -1):
|
|
reply = wp.decide(req(kind), cfg())
|
|
self.assertEqual(reply["action"], "reject", "kind %d" % kind)
|
|
self.assertIn("kind not accepted", reply["msg"])
|
|
|
|
def test_marketplace_kind_accepted_and_zap_rejected(self):
|
|
# A newly-allowed Magick Market kind (NIP-89 handler info) is accepted.
|
|
self.assertEqual(wp.decide(req(31990), cfg())["action"], "accept")
|
|
# A still-rejected kind (Lightning zap receipt, disabled in the
|
|
# GRIN-only marketplace) is refused by the default-deny whitelist.
|
|
reply = wp.decide(req(9735), cfg())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("kind not accepted", reply["msg"])
|
|
|
|
def test_malformed_kind_fails_closed(self):
|
|
for bad in (None, "1059", 3.5, True, [1059]):
|
|
r = req(0)
|
|
r["event"]["kind"] = bad
|
|
self.assertEqual(wp.decide(r, cfg())["action"], "reject", repr(bad))
|
|
|
|
def test_missing_or_bad_event_fails_closed(self):
|
|
self.assertEqual(wp.decide({"type": "new"}, cfg())["action"], "reject")
|
|
self.assertEqual(wp.decide({"event": "nope"}, cfg())["action"], "reject")
|
|
|
|
def test_custom_kind_list_env(self):
|
|
# Kind 1 is author-locked, so authorize PK to see the whitelist pass.
|
|
c = wp.load_config(
|
|
env={"FLOONET_ALLOWED_KINDS": "1,7", "FLOONET_AUTHORIZED_AUTHORS": PK}
|
|
)
|
|
self.assertEqual(wp.decide(req(1), c)["action"], "accept")
|
|
self.assertEqual(wp.decide(req(0), c)["action"], "reject")
|
|
|
|
def test_empty_or_garbage_kind_list_refused_at_startup(self):
|
|
with self.assertRaises(SystemExit):
|
|
wp.load_config(env={"FLOONET_ALLOWED_KINDS": ""})
|
|
with self.assertRaises(SystemExit):
|
|
wp.load_config(env={"FLOONET_ALLOWED_KINDS": "0,x"})
|
|
|
|
|
|
class AuthRequirement(unittest.TestCase):
|
|
def test_off_by_default(self):
|
|
self.assertEqual(wp.decide(req(1059), cfg())["action"], "accept")
|
|
|
|
def test_unauthed_rejected_when_required(self):
|
|
c = cfg(require_auth=True)
|
|
reply = wp.decide(req(1059), c)
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("auth-required", reply["msg"])
|
|
|
|
def test_authed_accepted_when_required(self):
|
|
c = cfg(require_auth=True)
|
|
self.assertEqual(wp.decide(req(1059, authed=PK), c)["action"], "accept")
|
|
|
|
def test_malformed_authed_rejected(self):
|
|
c = cfg(require_auth=True)
|
|
self.assertEqual(wp.decide(req(1059, authed="short"), c)["action"], "reject")
|
|
|
|
|
|
class _Authority(BaseHTTPRequestHandler):
|
|
"""Stub name authority: /api/v1/paid/<paid-pubkey> answers paid."""
|
|
|
|
paid_pubkeys = set()
|
|
fail = False
|
|
|
|
def do_GET(self):
|
|
if _Authority.fail:
|
|
self.send_response(500)
|
|
self.end_headers()
|
|
return
|
|
pk = self.path.rsplit("/", 1)[-1]
|
|
body = json.dumps({"pubkey": pk, "paid": pk in _Authority.paid_pubkeys})
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(body.encode())
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
|
|
class PaidWriteGate(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.server = HTTPServer(("127.0.0.1", 0), _Authority)
|
|
threading.Thread(target=cls.server.serve_forever, daemon=True).start()
|
|
cls.url = "http://127.0.0.1:%d" % cls.server.server_port
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
|
|
def setUp(self):
|
|
wp._paid_cache.clear()
|
|
_Authority.paid_pubkeys = set()
|
|
_Authority.fail = False
|
|
|
|
def c(self):
|
|
return cfg(pay_mode="write", authority_url=self.url, paid_cache_secs=60.0)
|
|
|
|
def test_unauthed_rejected_in_write_mode(self):
|
|
reply = wp.decide(req(1059), self.c())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("auth-required", reply["msg"])
|
|
|
|
def test_unpaid_pubkey_rejected(self):
|
|
reply = wp.decide(req(1059, authed=PK), self.c())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("payment required", reply["msg"])
|
|
|
|
def test_paid_pubkey_accepted(self):
|
|
_Authority.paid_pubkeys = {PK}
|
|
self.assertEqual(wp.decide(req(1059, authed=PK), self.c())["action"], "accept")
|
|
|
|
def test_verdict_cached_within_ttl(self):
|
|
_Authority.paid_pubkeys = {PK}
|
|
c = self.c()
|
|
self.assertEqual(wp.decide(req(1059, authed=PK), c)["action"], "accept")
|
|
# Authority flips to unpaid, but the cached verdict still applies.
|
|
_Authority.paid_pubkeys = set()
|
|
self.assertEqual(wp.decide(req(1059, authed=PK), c)["action"], "accept")
|
|
# Once the cache expires the fresh (unpaid) verdict is used.
|
|
wp._paid_cache[PK] = (True, 0.0)
|
|
self.assertEqual(wp.decide(req(1059, authed=PK), c)["action"], "reject")
|
|
|
|
def test_authority_down_fails_closed(self):
|
|
_Authority.fail = True
|
|
reply = wp.decide(req(1059, authed=PK), self.c())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("payment status unavailable", reply["msg"])
|
|
|
|
def test_kind_check_still_first_in_write_mode(self):
|
|
_Authority.paid_pubkeys = {PK}
|
|
reply = wp.decide(req(9735, authed=PK), self.c())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("kind not accepted", reply["msg"])
|
|
|
|
|
|
class PublicNoteLock(unittest.TestCase):
|
|
"""The public-note lockdown: kinds 1 and 30023 are accepted only from
|
|
operator-authorized authors; everything else is unaffected."""
|
|
|
|
def test_locked_kind_from_unauthorized_key_rejected(self):
|
|
c = cfg(authorized_authors={PK})
|
|
for kind in (1, 30023):
|
|
reply = wp.decide(req(kind, pubkey=OTHER_PK), c)
|
|
self.assertEqual(reply["action"], "reject", "kind %d" % kind)
|
|
self.assertIn("authorized authors", reply["msg"])
|
|
|
|
def test_locked_kind_from_authorized_hex_key_accepted(self):
|
|
c = wp.load_config(env={"FLOONET_AUTHORIZED_AUTHORS": PK})
|
|
for kind in (1, 30023):
|
|
self.assertEqual(wp.decide(req(kind), c)["action"], "accept", "kind %d" % kind)
|
|
|
|
def test_locked_kind_from_authorized_npub_accepted(self):
|
|
# Same key, configured as an npub instead of hex.
|
|
c = wp.load_config(env={"FLOONET_AUTHORIZED_AUTHORS": PK_NPUB})
|
|
self.assertEqual(c["authorized_authors"], {PK})
|
|
for kind in (1, 30023):
|
|
self.assertEqual(wp.decide(req(kind), c)["action"], "accept", "kind %d" % kind)
|
|
|
|
def test_non_locked_kinds_unaffected_by_random_keys(self):
|
|
# No authors configured; profiles, gift wraps and marketplace listings
|
|
# from arbitrary keys are still accepted (kind 0 must stay open).
|
|
c = cfg()
|
|
self.assertEqual(c["authorized_authors"], set())
|
|
for kind in (0, 1059, 30402):
|
|
self.assertEqual(
|
|
wp.decide(req(kind, pubkey=OTHER_PK), c)["action"], "accept", "kind %d" % kind
|
|
)
|
|
|
|
def test_closed_by_default_when_no_authors(self):
|
|
c = cfg()
|
|
for kind in (1, 30023):
|
|
reply = wp.decide(req(kind), c)
|
|
self.assertEqual(reply["action"], "reject", "kind %d" % kind)
|
|
self.assertIn("authorized authors", reply["msg"])
|
|
|
|
def test_malformed_npub_skipped_without_crash(self):
|
|
# A garbage npub, a too-short hex, and a good hex in one list: the
|
|
# good one survives, the bad ones are dropped, the plugin lives.
|
|
c = wp.load_config(
|
|
env={"FLOONET_AUTHORIZED_AUTHORS": "npub1notvalid,dead,%s" % PK}
|
|
)
|
|
self.assertEqual(c["authorized_authors"], {PK})
|
|
self.assertEqual(wp.decide(req(1), c)["action"], "accept")
|
|
|
|
def test_mixed_hex_and_npub_and_whitespace(self):
|
|
c = wp.load_config(
|
|
env={"FLOONET_AUTHORIZED_AUTHORS": " %s , %s " % (PK_NPUB, OTHER_PK)}
|
|
)
|
|
self.assertEqual(c["authorized_authors"], {PK, OTHER_PK})
|
|
|
|
|
|
class GiftWrapRetention(unittest.TestCase):
|
|
"""Kind 1059 (gift wrap) retention/shape guard: no NIP-40 expiration tag
|
|
(strfry's only automatic deletion trigger), exactly one well-formed `p`
|
|
recipient tag. Every other kind is unaffected."""
|
|
|
|
def test_expiration_tag_rejected(self):
|
|
r = req(1059, tags=[["p", OTHER_PK], ["expiration", "1700000100"]])
|
|
reply = wp.decide(r, cfg())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("expiration not allowed", reply["msg"])
|
|
|
|
def test_no_p_tag_rejected(self):
|
|
reply = wp.decide(req(1059, tags=[]), cfg())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("missing recipient", reply["msg"])
|
|
|
|
def test_multiple_p_tags_rejected(self):
|
|
r = req(1059, tags=[["p", OTHER_PK], ["p", PK]])
|
|
reply = wp.decide(r, cfg())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("missing recipient", reply["msg"])
|
|
|
|
def test_malformed_p_tag_rejected(self):
|
|
for bad in ("short", "z" * 64, "", None):
|
|
r = req(1059, tags=[["p", bad]])
|
|
reply = wp.decide(r, cfg())
|
|
self.assertEqual(reply["action"], "reject", repr(bad))
|
|
self.assertIn("missing recipient", reply["msg"])
|
|
|
|
def test_valid_giftwrap_accepted(self):
|
|
r = req(1059, tags=[["p", OTHER_PK]])
|
|
self.assertEqual(wp.decide(r, cfg())["action"], "accept")
|
|
|
|
def test_non_giftwrap_with_expiration_unaffected(self):
|
|
# Expiration tags are fine on every other kind; only 1059 is guarded.
|
|
r = req(1, tags=[["expiration", "1700000100"]])
|
|
reply = wp.decide(r, cfg(authorized_authors={PK}))
|
|
self.assertEqual(reply["action"], "accept")
|
|
|
|
def test_giftwrap_guard_runs_before_paid_gate(self):
|
|
# Malformed gift-wrap shape is caught even when the paid gate would
|
|
# otherwise reject/accept on auth; shape check must run first.
|
|
c = cfg(pay_mode="write")
|
|
r = req(1059, authed=PK, tags=[["expiration", "1700000100"], ["p", OTHER_PK]])
|
|
reply = wp.decide(r, c)
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("expiration not allowed", reply["msg"])
|
|
|
|
def test_uppercase_p_tag_rejected(self):
|
|
# The recipient-only READ gate compares `#p` to the authed pubkey as
|
|
# a case-sensitive lowercase string; an uppercase `p` would parse as
|
|
# hex but never match, orphaning the payment forever. Reject it here.
|
|
r = req(1059, tags=[["p", OTHER_PK.upper()]])
|
|
reply = wp.decide(r, cfg())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("missing recipient", reply["msg"])
|
|
|
|
def test_mixed_case_p_tag_rejected(self):
|
|
mixed = "B" + OTHER_PK[1:]
|
|
r = req(1059, tags=[["p", mixed]])
|
|
reply = wp.decide(r, cfg())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("missing recipient", reply["msg"])
|
|
|
|
def test_valid_lowercase_p_tag_accepted(self):
|
|
r = req(1059, tags=[["p", OTHER_PK]])
|
|
self.assertEqual(wp.decide(r, cfg())["action"], "accept")
|
|
|
|
def test_minimal_valid_wrap_accepted(self):
|
|
# A spec-accurate NIP-59 gift wrap: exactly one tag, the `p`
|
|
# recipient, lowercase hex, nothing else.
|
|
r = req(1059, tags=[["p", OTHER_PK]])
|
|
reply = wp.decide(r, cfg())
|
|
self.assertEqual(reply["action"], "accept")
|
|
|
|
def test_extraneous_tags_rejected(self):
|
|
# A well-formed `p` tag plus unrelated junk tags: reject as bloat
|
|
# rather than let a 128 KiB event carry thousands of extra tags.
|
|
r = req(1059, tags=[["p", OTHER_PK], ["junk", "1"], ["e", PK]])
|
|
reply = wp.decide(r, cfg())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("extraneous tags", reply["msg"])
|
|
|
|
def test_many_extraneous_tags_rejected(self):
|
|
# A pathological case: one valid `p` tag buried under a pile of
|
|
# junk tags, well under the 128 KiB event-size ceiling.
|
|
tags = [["p", OTHER_PK]] + [["junk", str(i)] for i in range(2000)]
|
|
reply = wp.decide(req(1059, tags=tags), cfg())
|
|
self.assertEqual(reply["action"], "reject")
|
|
self.assertIn("extraneous tags", reply["msg"])
|
|
|
|
def test_non_giftwrap_with_many_tags_unaffected(self):
|
|
# The tag ceiling is 1059-only; other kinds are free to carry as
|
|
# many tags as they like.
|
|
tags = [["t", str(i)] for i in range(50)]
|
|
reply = wp.decide(req(1, tags=tags), cfg(authorized_authors={PK}))
|
|
self.assertEqual(reply["action"], "accept")
|
|
|
|
|
|
class EnvFileConfig(unittest.TestCase):
|
|
"""load_config also reads a KEY=VALUE file, with real env taking priority."""
|
|
|
|
def _write(self, body):
|
|
import tempfile
|
|
fd, path = tempfile.mkstemp(prefix="floonet-env-")
|
|
with os.fdopen(fd, "w") as fh:
|
|
fh.write(body)
|
|
self.addCleanup(os.remove, path)
|
|
return path
|
|
|
|
def test_env_file_supplies_config(self):
|
|
path = self._write(
|
|
"# floonet config\n"
|
|
"FLOONET_ALLOWED_KINDS = 1,7\n"
|
|
"\n"
|
|
"FLOONET_AUTHORIZED_AUTHORS=%s\n" % PK
|
|
)
|
|
c = wp.load_config(env={"FLOONET_ENV_FILE": path})
|
|
self.assertEqual(c["allowed_kinds"], frozenset({1, 7}))
|
|
self.assertEqual(c["authorized_authors"], {PK})
|
|
self.assertEqual(wp.decide(req(1), c)["action"], "accept")
|
|
self.assertEqual(wp.decide(req(0), c)["action"], "reject")
|
|
|
|
def test_real_env_overrides_file(self):
|
|
path = self._write("FLOONET_ALLOWED_KINDS=1\n")
|
|
c = wp.load_config(
|
|
env={"FLOONET_ENV_FILE": path, "FLOONET_ALLOWED_KINDS": "7"}
|
|
)
|
|
self.assertEqual(c["allowed_kinds"], frozenset({7}))
|
|
|
|
def test_missing_file_is_harmless(self):
|
|
c = wp.load_config(env={"FLOONET_ENV_FILE": "/no/such/floonet.env"})
|
|
self.assertIn(1059, c["allowed_kinds"])
|
|
|
|
|
|
class Bech32Decoder(unittest.TestCase):
|
|
def test_known_npub_decodes(self):
|
|
self.assertEqual(wp._npub_to_hex(PK_NPUB), PK)
|
|
|
|
def test_invalid_npubs_return_none(self):
|
|
for bad in (
|
|
"npub1notvalid",
|
|
"npub1424242424242424242424242424242424242424242424242424qamrcaX", # bad checksum
|
|
"nsec1qqqqq", # wrong hrp
|
|
"",
|
|
"424242",
|
|
):
|
|
self.assertIsNone(wp._npub_to_hex(bad), bad)
|
|
|
|
|
|
class StrfryPipeProtocol(unittest.TestCase):
|
|
"""Run the plugin as strfry does: one JSONL request per line on stdin,
|
|
one JSONL reply per line on stdout, in order."""
|
|
|
|
def run_plugin(self, lines, env=None):
|
|
e = {"PATH": os.environ.get("PATH", ""), "FLOONET_PAY_MODE": "off"}
|
|
if env:
|
|
e.update(env)
|
|
proc = subprocess.run(
|
|
[sys.executable, PLUGIN],
|
|
input="".join(json.dumps(l) + "\n" for l in lines),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
env=e,
|
|
)
|
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
|
return [json.loads(out) for out in proc.stdout.splitlines()]
|
|
|
|
def test_accept_and_reject_over_the_wire(self):
|
|
replies = self.run_plugin([req(1059, event_id="ok1"), req(9735, event_id="no1"), req(0, event_id="ok2")])
|
|
self.assertEqual(
|
|
[(r["id"], r["action"]) for r in replies],
|
|
[("ok1", "accept"), ("no1", "reject"), ("ok2", "reject" if 0 not in DEFAULT_KINDS else "accept")],
|
|
)
|
|
|
|
def test_malformed_line_fails_closed_and_loop_survives(self):
|
|
proc = subprocess.run(
|
|
[sys.executable, PLUGIN],
|
|
input="this is not json\n" + json.dumps(req(1059, event_id="after")) + "\n",
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
env={"PATH": os.environ.get("PATH", "")},
|
|
)
|
|
self.assertEqual(proc.returncode, 0)
|
|
replies = [json.loads(out) for out in proc.stdout.splitlines()]
|
|
self.assertEqual(replies[0]["action"], "reject")
|
|
self.assertEqual((replies[1]["id"], replies[1]["action"]), ("after", "accept"))
|
|
|
|
def test_env_whitelist_respected_over_the_wire(self):
|
|
replies = self.run_plugin(
|
|
[req(1, event_id="now-ok")],
|
|
env={"FLOONET_ALLOWED_KINDS": "1", "FLOONET_AUTHORIZED_AUTHORS": PK},
|
|
)
|
|
self.assertEqual((replies[0]["id"], replies[0]["action"]), ("now-ok", "accept"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|