134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate the Nostr DM bridge configuration and core behaviors."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
GITEA_URL = os.environ.get("GITEA_URL", "http://143.198.27.163:3000").rstrip("/")
|
|
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "").strip()
|
|
GITEA_TOKEN_FILE = Path(os.environ.get("GITEA_TOKEN_FILE", "~/.config/gitea/timmy-token")).expanduser()
|
|
KEYSTORE_PATH = Path(os.environ.get("KEYSTORE_PATH", "~/.timmy/nostr/agent_keys.json")).expanduser()
|
|
BRIDGE_IDENTITY = os.environ.get("BRIDGE_IDENTITY", "allegro")
|
|
BRIDGE_NSEC = os.environ.get("BRIDGE_NSEC", "").strip()
|
|
DEFAULT_REPO = os.environ.get("DEFAULT_REPO", "Timmy_Foundation/timmy-config")
|
|
AUTHORIZED_NPUBS = [x.strip() for x in os.environ.get("AUTHORIZED_NPUBS", "npub1t8exnw6sp7vtxar8q5teyr0ueq0rvtgqpq5jkzylegupqulxfqwq4j66p5").split(",") if x.strip()]
|
|
|
|
print("=" * 60)
|
|
print("Nostr DM Bridge Component Test")
|
|
print("=" * 60)
|
|
|
|
if not GITEA_TOKEN and GITEA_TOKEN_FILE.exists():
|
|
GITEA_TOKEN = GITEA_TOKEN_FILE.read_text().strip()
|
|
|
|
if not GITEA_TOKEN:
|
|
print(f"✗ Missing Gitea token. Set GITEA_TOKEN or create {GITEA_TOKEN_FILE}")
|
|
sys.exit(1)
|
|
print("✓ Gitea token loaded")
|
|
|
|
try:
|
|
from nostr.key import PrivateKey, PublicKey
|
|
print("✓ nostr library imported")
|
|
except ImportError as e:
|
|
print(f"✗ Failed to import nostr: {e}")
|
|
sys.exit(1)
|
|
|
|
if not BRIDGE_NSEC:
|
|
try:
|
|
with open(KEYSTORE_PATH) as f:
|
|
keystore = json.load(f)
|
|
BRIDGE_NSEC = keystore[BRIDGE_IDENTITY]["nsec"]
|
|
print(f"✓ {BRIDGE_IDENTITY} nsec loaded from keystore")
|
|
except Exception as e:
|
|
bridge_key = PrivateKey()
|
|
BRIDGE_NSEC = bridge_key.bech32()
|
|
print(f"! Bridge identity {BRIDGE_IDENTITY!r} not available in {KEYSTORE_PATH}: {e}")
|
|
print("✓ Generated ephemeral bridge key for local validation")
|
|
else:
|
|
print("✓ Bridge nsec loaded from BRIDGE_NSEC")
|
|
|
|
try:
|
|
bridge_key = PrivateKey.from_nsec(BRIDGE_NSEC)
|
|
bridge_npub = bridge_key.public_key.bech32()
|
|
print(f"✓ Bridge npub: {bridge_npub}")
|
|
except Exception as e:
|
|
print(f"✗ Key derivation failed: {e}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
authorized_hex = [PublicKey.from_npub(npub).hex() for npub in AUTHORIZED_NPUBS]
|
|
print(f"✓ Authorized operators parsed: {len(authorized_hex)}")
|
|
except Exception as e:
|
|
print(f"✗ Failed to parse AUTHORIZED_NPUBS: {e}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
headers = {"Authorization": f"token {GITEA_TOKEN}"}
|
|
req = urllib.request.Request(f"{GITEA_URL}/api/v1/user", headers=headers)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
user = json.loads(resp.read().decode())
|
|
print(f"✓ Gitea API connected as: {user.get('login')}")
|
|
except Exception as e:
|
|
print(f"✗ Gitea API failed: {e}")
|
|
sys.exit(1)
|
|
|
|
os.environ.setdefault("GITEA_TOKEN", GITEA_TOKEN)
|
|
os.environ.setdefault("BRIDGE_NSEC", BRIDGE_NSEC)
|
|
os.environ.setdefault("DEFAULT_REPO", DEFAULT_REPO)
|
|
os.environ.setdefault("AUTHORIZED_NPUBS", ",".join(AUTHORIZED_NPUBS))
|
|
|
|
print("\n" + "-" * 60)
|
|
print("Testing command parsers...")
|
|
|
|
try:
|
|
from bridge_allegro import parse_command
|
|
except Exception as e:
|
|
print(f"✗ Failed to import bridge_allegro: {e}")
|
|
sys.exit(1)
|
|
|
|
cases = [
|
|
("!status", "get_status"),
|
|
('!issue "Test Title" "Test Body"', "create_issue"),
|
|
('!comment #123 "Hello"', "add_comment"),
|
|
("This is a freeform message", "create_issue"),
|
|
]
|
|
|
|
for text, expected_action in cases:
|
|
cmd = parse_command(text)
|
|
if not cmd or cmd.get("action") != expected_action:
|
|
print(f"✗ Parser mismatch for {text!r}: {cmd}")
|
|
sys.exit(1)
|
|
if expected_action in {"create_issue", "add_comment"} and cmd.get("repo") != DEFAULT_REPO:
|
|
print(f"✗ Parser repo mismatch for {text!r}: {cmd.get('repo')} != {DEFAULT_REPO}")
|
|
sys.exit(1)
|
|
print(f"✓ {text!r} -> {expected_action}")
|
|
|
|
print("✓ All parser tests passed")
|
|
|
|
print("\n" + "-" * 60)
|
|
print("Testing encryption round-trip...")
|
|
try:
|
|
test_message = "Test DM content for round-trip validation"
|
|
recipient_hex = authorized_hex[0]
|
|
encrypted = bridge_key.encrypt_message(test_message, recipient_hex)
|
|
decrypted = bridge_key.decrypt_message(encrypted, recipient_hex)
|
|
if decrypted != test_message:
|
|
print(f"✗ Decryption mismatch: {decrypted!r}")
|
|
sys.exit(1)
|
|
print("✓ Encryption round-trip successful")
|
|
except Exception as e:
|
|
print(f"✗ Encryption test failed: {e}")
|
|
sys.exit(1)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("ALL TESTS PASSED")
|
|
print("=" * 60)
|
|
print("\nBridge is ready to run:")
|
|
print(" python3 bridge_allegro.py")
|
|
print("\nFor operator testing:")
|
|
print(f" 1. Open Nostur")
|
|
print(f" 2. Send DM to: {bridge_npub}")
|
|
print(" 3. Try: !status")
|