fix(buzz): normalize npub entries in BUZZ_ALLOWED_USERS to hex

Summary:
The gateway's central allowlist check compared the inbound Buzz sender's
64-char hex pubkey against the raw BUZZ_ALLOWED_USERS entries. An operator
who listed only their npub saw every message rejected with
"Unauthorized user: <hex pubkey>" (gateway drops the message). npub
entries are now decoded to hex before the comparison, so npub and hex
forms of the same identity are equivalent.

Root Cause:
The Buzz adapter's own intake check already normalizes npub→hex via
_normalize_user_ref when building _allowed_pubkeys, but the gateway
applies BUZZ_ALLOWED_USERS centrally as well (authz_mixin._is_user_authorized
via the platform registry's allowed_users_env). That central path did a
raw string comparison of the allowlist entries against the hex user_id,
so an npub-only entry never matched.

Change:
- gateway/authz_mixin.py: add a pure-stdlib bech32 npub→hex decoder
  (mirroring plugins/platforms/buzz/adapter.py) and normalize the buzz
  allowlist set in _is_user_authorized: each npub1… entry is decoded and
  its hex form added; hex entries pass through unchanged, so existing
  hex-only allowlists keep working. Comparison stays fail-closed for
  unrelated senders.
- tests/gateway/test_buzz_authz.py: new tests covering npub-only,
  hex-only, mixed, uppercase-npub, and denial of unrelated users, plus
  unit tests for the decoder/helper.

Verification:
- 43 passed (test_buzz_authz.py + test_buzz_adapter.py +
  test_pairing_allowlist_bypass.py)
- 9 passed (test_multiplex_profile_authz.py + test_buzz_websocket.py)

Closes #78428
This commit is contained in:
webtecnica
2026-08-04 08:18:11 -03:00
committed by Teknium
parent a5e7355766
commit c83121cf5f
2 changed files with 222 additions and 0 deletions

View File

@@ -104,6 +104,91 @@ def _coerce_allow_set(raw) -> set[str]:
return {part.strip() for part in str(raw).split(",") if part.strip()}
# ---------------------------------------------------------------------------
# Nostr npub → hex normalization (Buzz and future Nostr-based platforms).
#
# ``BUZZ_ALLOWED_USERS`` accepts either a 64-char hex pubkey or an ``npub1…``
# bech32 string, but inbound event pubkeys are always hex. Without decoding,
# the central allowlist comparison string-matches the raw npub against the
# hex pubkey and an operator who listed only their npub sees every message
# rejected ("Unauthorized user: <hex pubkey>", #78428). Pure stdlib; mirrors
# the decoder in plugins/platforms/buzz/adapter.py.
# ---------------------------------------------------------------------------
_BECH32_CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
def _bech32_polymod(values):
chk = 1
generator = [0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3]
for value in values:
top = chk >> 25
chk = (chk & 0x1FFFFFF) << 5 ^ value
for i in range(5):
chk ^= generator[i] if ((top >> i) & 1) else 0
return chk
def _bech32_hrp_expand(hrp: str):
return [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp]
def _convertbits(data, frombits: int, tobits: int, pad: bool = True):
acc = 0
bits = 0
ret = []
maxv = (1 << tobits) - 1
for value in data:
if value < 0 or (value >> frombits):
return None
acc = (acc << frombits) | value
bits += frombits
while bits >= tobits:
bits -= tobits
ret.append((acc >> bits) & maxv)
if pad:
if bits:
ret.append((acc << (tobits - bits)) & maxv)
elif bits >= frombits or ((acc << (tobits - bits)) & maxv):
return None
return ret
def _npub_to_hex(npub: str) -> Optional[str]:
"""Decode an ``npub1…`` bech32 string to a 64-char hex pubkey, else None."""
npub = npub.strip().lower()
if not npub.startswith("npub1"):
return None
data_part = npub[len("npub1"):]
try:
data = [_BECH32_CHARSET.index(c) for c in data_part]
except ValueError:
return None
if _bech32_polymod(_bech32_hrp_expand("npub") + data) != 1:
return None
decoded = _convertbits(data[:-6], 5, 8, pad=False)
if decoded is None or len(decoded) != 32:
return None
return bytes(decoded).hex()
def _normalize_nostr_allow_entries(entries: set) -> set:
"""Expand npub entries in an allowlist set to their hex pubkey form.
Hex entries pass through unchanged; each valid ``npub1…`` entry is decoded
and its 64-char hex form added, so either form authorizes the same
identity (#78428). Invalid entries are kept as-is (they simply never
match an inbound hex pubkey).
"""
expanded = set(entries)
for entry in entries:
if entry.lower().startswith("npub1"):
hex_key = _npub_to_hex(entry)
if hex_key:
expanded.add(hex_key)
return expanded
class GatewayAuthorizationMixin:
"""User/chat authorization methods for ``GatewayRunner``."""
@@ -864,6 +949,18 @@ class GatewayAuthorizationMixin:
):
check_ids.add(source.user_name)
# Buzz (Nostr-based): BUZZ_ALLOWED_USERS accepts npub or hex, but
# inbound event pubkeys are always 64-char hex. Decode npub entries
# to hex so an operator who listed only their npub authorizes the
# same identity as the hex form (#78428). Hex entries pass through
# unchanged, so existing hex-only allowlists keep working.
if source.platform is not None and source.platform.value == "buzz":
allowed_ids = _normalize_nostr_allow_entries(allowed_ids)
if user_id.startswith("npub"):
hex_user = _npub_to_hex(user_id)
if hex_user:
check_ids.add(hex_user)
return bool(check_ids & allowed_ids)
def _get_unauthorized_dm_behavior(

View File

@@ -0,0 +1,125 @@
"""Gateway authz tests: BUZZ_ALLOWED_USERS accepts npub or hex (#78428).
Inbound Buzz events carry the sender as a 64-char hex pubkey, while
``BUZZ_ALLOWED_USERS`` historically accepted npubs too. The gateway's
central allowlist comparison must decode npub entries to hex at comparison
time, or an operator who listed only their npub is rejected with
"Unauthorized user: <hex pubkey>" (gateway drops the message).
"""
import pytest
from gateway.config import Platform
from gateway.platform_registry import PlatformEntry, platform_registry
from gateway.session import SessionSource
# Chip's public identity (public information, not a secret) — the same pair
# used by tests/gateway/test_buzz_adapter.py.
SELF_PUBKEY = "9fd5c7ba6d3ef224da78f541e0fcb9c50f72cc63edb19aae76ac6a0474dfa860"
SELF_NPUB = "npub1nl2u0wnd8mezfknc74q7pl9ec58h9nrrakce4tnk434qgaxl4psqe5twr6"
OTHER_PUBKEY = "a" * 64
_BUZZ_PLATFORM = Platform("buzz")
_AUTH_ENV_VARS = (
"BUZZ_ALLOWED_USERS",
"BUZZ_ALLOW_ALL_USERS",
"GATEWAY_ALLOWED_USERS",
"GATEWAY_ALLOW_ALL_USERS",
)
@pytest.fixture(autouse=True)
def _isolate_env(monkeypatch):
for var in _AUTH_ENV_VARS:
monkeypatch.delenv(var, raising=False)
@pytest.fixture
def buzz_registered():
"""Register a minimal Buzz platform entry (allowed_users_env contract)."""
platform_registry.register(
PlatformEntry(
name="buzz",
label="Buzz",
adapter_factory=lambda cfg: None,
check_fn=lambda: True,
allowed_users_env="BUZZ_ALLOWED_USERS",
allow_all_env="BUZZ_ALLOW_ALL_USERS",
)
)
yield
platform_registry.unregister("buzz")
def _make_runner():
from gateway.run import GatewayRunner
runner = object.__new__(GatewayRunner)
runner.pairing_store = None
return runner
def _make_source(user_id: str, chat_type: str = "dm"):
return SessionSource(
platform=_BUZZ_PLATFORM,
chat_id="ccc2bc1a-7a82-5a8f-8c4e-57a070cbe7cd",
chat_type=chat_type,
user_id=user_id,
user_name="Chip",
is_bot=False,
)
def test_npub_only_allowlist_authorizes_hex_identity(monkeypatch, buzz_registered):
"""The reported bug: listing only the npub must authorize the hex pubkey."""
monkeypatch.setenv("BUZZ_ALLOWED_USERS", SELF_NPUB)
runner = _make_runner()
assert runner._is_user_authorized(_make_source(SELF_PUBKEY)) is True
def test_hex_only_allowlist_still_authorizes(monkeypatch, buzz_registered):
"""Existing hex-only allowlists keep working unchanged."""
monkeypatch.setenv("BUZZ_ALLOWED_USERS", SELF_PUBKEY)
runner = _make_runner()
assert runner._is_user_authorized(_make_source(SELF_PUBKEY)) is True
def test_mixed_npub_and_hex_allowlist_authorizes(monkeypatch, buzz_registered):
"""Both forms in one allowlist authorize the same identity."""
monkeypatch.setenv("BUZZ_ALLOWED_USERS", f"{SELF_NPUB},{SELF_PUBKEY}")
runner = _make_runner()
assert runner._is_user_authorized(_make_source(SELF_PUBKEY)) is True
def test_npub_allowlist_still_denies_other_user(monkeypatch, buzz_registered):
"""Normalization must not turn into fail-open for unrelated senders."""
monkeypatch.setenv("BUZZ_ALLOWED_USERS", SELF_NPUB)
runner = _make_runner()
assert runner._is_user_authorized(_make_source(OTHER_PUBKEY)) is False
def test_uppercase_npub_allowlist_authorizes(monkeypatch, buzz_registered):
"""npub entries are case-insensitive, like the adapter's own decoder."""
monkeypatch.setenv("BUZZ_ALLOWED_USERS", SELF_NPUB.upper())
runner = _make_runner()
assert runner._is_user_authorized(_make_source(SELF_PUBKEY)) is True
def test_normalize_nostr_allow_entries():
from gateway.authz_mixin import _normalize_nostr_allow_entries
expanded = _normalize_nostr_allow_entries({SELF_NPUB, OTHER_PUBKEY, "not-a-key"})
assert SELF_PUBKEY in expanded
assert SELF_NPUB in expanded # original kept, harmless
assert OTHER_PUBKEY in expanded
assert "not-a-key" in expanded
def test_npub_to_hex_roundtrip():
from gateway.authz_mixin import _npub_to_hex
assert _npub_to_hex(SELF_NPUB) == SELF_PUBKEY
assert _npub_to_hex(SELF_PUBKEY) is None # hex input is not an npub
assert _npub_to_hex("npub1garbage!!") is None # invalid bech32
assert _npub_to_hex("") is None