-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathotrv4plus_xmpp.py
More file actions
2131 lines (1917 loc) · 80.4 KB
/
Copy pathotrv4plus_xmpp.py
File metadata and controls
2131 lines (1917 loc) · 80.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""
OTRv4+ XMPP - full OTR + SMP over XMPP, transported over I2P SAM
================================================================
Version: 10.10.4
Post-quantum OTRv4+ end-to-end encryption over XMPP, reusing the IRC client's
Rust-backed OTR engine (EnhancedSessionManager) unchanged. Every OTR frame
("?OTRv4 <base64>") rides in a single <message><body>, so there is NO
fragmentation and the post-quantum handshake/SMP are far faster than on IRC.
slixmpp (Python) -> "?OTRv4 ..." frames only, never keys
v
EnhancedSessionManager (Python) -> drives DAKE/SMP/ratchet
v
otrv4_core.so (Rust) -> ALL secrets, ZeroizeOnDrop, never exposed
TRANSPORT: I2P SAM (matching the IRC client). A SAM stream is opened to the
server's c2s .b32.i2p destination and exposed as a local TCP endpoint; slixmpp
connects to that endpoint and does STARTTLS normally, unaware of I2P. A
keepalive task pings the stream so idle I2P tunnels don't get torn down.
XEP SUPPORT (slixmpp plugins):
XEP-0030 Service discovery (capability advertisement)
XEP-0085 Chat state notifications
XEP-0115 Entity capabilities
XEP-0184 Message delivery receipts (auto mode)
XEP-0198 Stream management (stanza acks; graceful degradation if unsupported)
XEP-0199 XMPP Ping (peer reachability check via /ping)
POST-DAKE FLOW (identical to the IRC client):
1. DAKE completes -> session ENCRYPTED.
2. Both fingerprints are shown; you are asked "Trust this fingerprint? y/n".
- y -> fingerprint pinned as VERIFIED (TOFU trust DB).
- n -> encrypted-only.
3. You are prompted for the Socialist Millionaire Protocol passphrase, which
is stored for AUTO-RESPOND. Press Enter / "skip" to skip.
4. Once BOTH sides have stored the passphrase, EITHER side runs /smp start.
ROSTER / SUBSCRIPTION FLOW:
Subscription requests are NEVER auto-approved. They queue in /pending; use
/accept <jid> or /deny <jid> to respond. /add <jid> sends your own request,
/remove <jid> deletes a contact, /roster lists all contacts.
USAGE:
pip install slixmpp aiodns
python otrv4plus_xmpp.py \
--jid alice@<vhost>.b32.i2p \
--server <c2s-tunnel>.b32.i2p \
--peer bob@<vhost>.b32.i2p \
--insecure-tls --debug
COMMANDS:
/otr [jid] start an OTR session (DAKE)
y / n answer the trust-fingerprint prompt
<passphrase> answer the SMP passphrase prompt
/smp start begin SMP verification
/smp <secret> store a secret AND immediately start SMP
/smp-secret <secret> store a secret for auto-respond (no start)
/trust re-show fingerprints and the trust prompt
/msg <jid> <text> send plaintext (no OTR)
/status show session + trust + SMP state for --peer
/roster list all roster contacts
/add <jid> add a contact and send subscription request
/remove <jid> remove a contact from roster
/pending show pending subscription requests
/accept <jid> accept a pending subscription request
/deny <jid> deny a pending subscription request
/block <jid> block inbound messages from a JID (session-local)
/unblock <jid> remove a session-local block
/blocked list session-local blocks
/ping <jid> XMPP ping a peer (XEP-0199)
/help show this command list
/quit disconnect and exit
<text> send to --peer (auto-encrypts once OTR is up)
"""
# =============================================================================
# SECURITY MODEL (enforced throughout this file):
# * All cryptography lives in the Rust core (otrv4_core) via the shared
# EnhancedSessionManager. This transport never holds key material and never
# implements a primitive; it moves "?OTRv4 <base64>" frames and renders UI.
# * Every piece of untrusted data shown on the terminal passes through
# _sanitise(), which strips ANSI/OSC/CSI escape sequences, C0/C1 controls,
# and newlines. This blocks terminal-title hijack and forged log lines.
# * Inbound message fragments are bounded (index range, fragment count, and a
# per-peer reassembly cap) before stitching, preventing memory-exhaustion
# DoS and out-of-range indexing.
# * TLS verification is on by default; only disabled behind --insecure-tls
# which is acceptable over I2P (.b32 destination is cryptographically
# authenticated) but warned against on clearnet.
# * Fingerprints are pinned on first use (TOFU); the trust prompt gates the
# transition to a VERIFIED session.
# * Subscription requests are NEVER auto-approved. auto_authorize and
# auto_subscribe are both set to False. All subscription requests queue in
# _pending_subscriptions and require explicit /accept or /deny.
# * Inbound messages are rate-limited per peer (20 msgs / 5 s) to prevent
# event-loop flooding from a hostile or misbehaving peer.
# * SMP secrets are validated for minimum length (8 chars) and maximum
# length (512 chars) before being passed to the Rust engine.
# * A session-local block list (/block, /unblock) drops all inbound messages
# from listed JIDs without processing or displaying them.
# * XEP-0198 stream management registered for stanza acks; degrades
# gracefully if the server does not support it.
# * XEP-0184 delivery receipts enabled (auto mode).
# * Automatic reconnection with exponential backoff re-establishes the I2P
# SAM tunnel before reconnecting slixmpp. Disabled on auth failure.
#
# Audited for: shell/command injection, escape-sequence injection, ReDoS,
# unsafe deserialisation, weak hashing, and insecure randomness. None present:
# no eval/exec, no pickle/marshal, no shell=True, no user-compiled regexes,
# no md5/sha1, and no use of the `random` module for any security decision.
# =============================================================================
import argparse
import asyncio
import builtins
import collections
import getpass
import logging
import os
import re
import sys
import time
from concurrent.futures import ThreadPoolExecutor
XMPP_VERSION = "10.10.4"
OTR_MODULE = "otrv4plus" # symlink -> otrv4+.py
try:
_otr = __import__(OTR_MODULE)
EnhancedSessionManager = _otr.EnhancedSessionManager
OTRConfig = _otr.OTRConfig
OTRTracer = getattr(_otr, "OTRTracer", None)
I2PSAMConnection = getattr(_otr, "I2PSAMConnection", None)
except Exception as e:
print(f"Could not import OTR engine from '{OTR_MODULE}': {e}", file=sys.stderr)
print(
"Ensure otrv4+.py, the otrv4plus.py symlink, and otrv4_core.so are "
"in this directory.",
file=sys.stderr,
)
sys.exit(1)
# ---------------------------------------------------------------------------
# Terminal UI - REUSES the engine's own ANSI TUI (PanelManager / Screen / raw
# line editor), the same one the IRC client uses.
# ---------------------------------------------------------------------------
_PanelManager = getattr(_otr, "PanelManager", None)
_Screen = getattr(_otr, "Screen", None)
_UIConstants = getattr(_otr, "UIConstants", None)
_setup_raw_mode = getattr(_otr, "_setup_raw_mode", None)
_restore_terminal = getattr(_otr, "_restore_terminal", None)
_read_one_char = getattr(_otr, "_read_one_char", None)
_handle_input_char = getattr(_otr, "_handle_input_char", None)
_set_prompt = getattr(_otr, "_set_prompt", None)
_colorize = getattr(_otr, "colorize", lambda s, c: s)
_EOF_SENTINEL = getattr(_otr, "_EOF_SENTINEL", object())
_TUI_AVAILABLE = all(
x is not None
for x in (
_PanelManager,
_Screen,
_UIConstants,
_setup_raw_mode,
_restore_terminal,
_read_one_char,
_handle_input_char,
_set_prompt,
)
)
_ACTIVE_TUI_CLIENT = None
# Full session transcript (written under --debug, deleted on clean exit).
_SESSION_LOG_FH = None
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
def _sanitise(text, max_len: int = 1024) -> str:
"""Strip ANSI/OSC/CSI escape sequences and control characters from
untrusted data (peer JIDs, plaintext bodies, decrypted OTR payloads)
before writing to the terminal."""
text = str(text)
text = re.sub(r"\x1b[P\]X^_][^\x07\x1b]*(?:\x07|\x1b\\)", "", text)
text = re.sub(r"\x1b\[[\x30-\x3f]*[\x20-\x2f]*[\x40-\x7e]", "", text)
text = re.sub(r"\x1b[\x20-\x2f][\x30-\x7e]", "", text)
text = re.sub(r"\x1b.", "", text, flags=re.DOTALL)
text = re.sub(r"[\x00-\x08\x0a-\x1f\x7f\x80-\x9f]", "", text)
return text[:max_len]
# Lines carrying actual message content are redacted from the on-disk
# transcript so cleartext bodies never touch disk.
_LOG_CONTENT_RE = re.compile(r"^(\[(?:otr|plain)\] <[^>]*>)\s(.*)$", re.DOTALL)
def _log_to_file(msg):
if _SESSION_LOG_FH is None:
return
try:
clean = _ANSI_RE.sub("", msg)
m = _LOG_CONTENT_RE.match(clean)
if m:
clean = f"{m.group(1)} <message body redacted: {len(m.group(2))} chars>"
ts = time.strftime("%Y-%m-%d %H:%M:%S")
_SESSION_LOG_FH.write(f"{ts} {clean}\n")
_SESSION_LOG_FH.flush()
except Exception:
pass
def print(*args, **kwargs): # noqa: A001 (intentional module-scope shadow)
c = _ACTIVE_TUI_CLIENT
sep = kwargs.get("sep", " ")
msg = sep.join(str(a) for a in args)
_log_to_file(msg)
if c is not None and getattr(c, "_tui_enabled", False):
try:
c._tui_route_output(msg)
return
except Exception:
pass
builtins.print(*args, **kwargs)
try:
from slixmpp import ClientXMPP
from slixmpp.exceptions import IqError, IqTimeout
except ImportError:
builtins.print("slixmpp not installed. Run: pip install slixmpp aiodns", file=sys.stderr)
sys.exit(1)
OTR_PREFIX = "?OTRv4 "
OTR_PREFIX_B = b"?OTRv4 "
# SMP passphrase length bounds enforced before passing to the Rust engine.
SMP_MIN_LEN = 8
SMP_MAX_LEN = 512
# Rate limiting: max inbound messages per peer per window.
_RATE_MAX = 20
_RATE_WINDOW = 5.0 # seconds
# Reconnect backoff constants.
_RECONNECT_BASE = 5 # seconds (initial delay)
_RECONNECT_MAX = 300 # seconds (5 min ceiling)
# Matches a bare JID for output panel routing.
_JID_PATTERN = re.compile(r"[A-Za-z0-9_.+-]+@[A-Za-z0-9_.-]+")
def _fmt_fp(fp: str) -> str:
"""Format a fingerprint as space-separated groups of 8 hex chars."""
if not fp or fp == "unavailable":
return fp or "unavailable"
clean = fp.upper().replace(" ", "")
return " ".join(clean[i : i + 8] for i in range(0, len(clean), 8))
# =============================================================================
# I2P SAM forwarder
# =============================================================================
async def start_i2p_sam_forwarder(
dest_b32: str, dest_port: int, sam_host: str = "127.0.0.1", sam_port: int = 7656
):
"""
Open an I2P SAM stream to `dest_b32` and expose it as a local TCP endpoint.
Returns (local_host, local_port). slixmpp connects to the local endpoint and
does STARTTLS normally; bytes are piped over the SAM stream to the I2P
destination. The SAM connection, local server, and writer are kept alive on
the loop so they are not garbage-collected.
"""
if I2PSAMConnection is None:
raise RuntimeError(
"I2PSAMConnection not available from the OTR module; "
"cannot use I2P SAM transport."
)
loop = asyncio.get_event_loop()
sam = I2PSAMConnection(sam_host=sam_host, sam_port=sam_port)
def _do_sam():
s = sam.connect(dest_b32)
s.setblocking(False)
return s
print(f"[i2p] opening SAM stream to {dest_b32} (a cold tunnel can take 30-90s)...")
sam_sock = await loop.run_in_executor(None, _do_sam)
print("[i2p] SAM stream established.")
sam_reader, sam_writer = await asyncio.open_connection(sock=sam_sock)
# I2P tunnels can drop a stream when a large message is written as one
# burst. We pace writes in small chunks to avoid the SAM cliff (~8KB).
SAM_CHUNK = 1024 # bytes per write toward I2P
SAM_CHUNK_DELAY = 0.02 # seconds between chunks on large messages
async def _handle_local(local_reader, local_writer):
async def pump_to_i2p(src, dst):
try:
while True:
data = await src.read(65536)
if not data:
break
if len(data) <= SAM_CHUNK:
dst.write(data)
await dst.drain()
else:
for i in range(0, len(data), SAM_CHUNK):
dst.write(data[i : i + SAM_CHUNK])
await dst.drain()
await asyncio.sleep(SAM_CHUNK_DELAY)
except Exception:
pass
finally:
try:
dst.close()
except Exception:
pass
async def pump_from_i2p(src, dst):
try:
while True:
data = await src.read(65536)
if not data:
break
dst.write(data)
await dst.drain()
except Exception:
pass
finally:
try:
dst.close()
except Exception:
pass
await asyncio.gather(
pump_to_i2p(local_reader, sam_writer),
pump_from_i2p(sam_reader, local_writer),
)
server = await asyncio.start_server(_handle_local, "127.0.0.1", 0)
host, port = server.sockets[0].getsockname()[:2]
if not hasattr(loop, "_i2p_keep"):
loop._i2p_keep = []
loop._i2p_keep.extend([sam, server, sam_writer])
print(f"[i2p] local bridge ready at {host}:{port} -> {dest_b32}")
return host, port
# =============================================================================
# XMPP client
# =============================================================================
class OTRv4PlusXMPP(ClientXMPP):
"""XMPP transport driving the OTRv4+ engine, with IRC-identical SMP flow."""
def __init__(self, jid, password, peer=None):
super().__init__(jid, password)
self.peer = peer
# Per-peer UI state.
self._pending = {} # peer -> 'trust' | 'smp_secret' | None
self._encrypted = set() # peers whose DAKE has completed
self._smp_reported = set() # (peer, state) already announced
self._frag_seq = 0 # monotonic id for outbound fragment sets
# Security: subscription approval queue; no auto-approval.
self._pending_subscriptions = {} # peer -> presence stanza
# Security: session-local block list.
self._blocked = set()
# Security: per-peer rate limiting.
self._rate_limit = {} # peer -> deque of timestamps
# Reconnect state (populated by main() before connect()).
self._sam_params = None # dict of SAM args for reconnect
self._is_i2p = False
self._shutting_down = False
self._reconnect_delay = _RECONNECT_BASE
self._reconnect_task = None
# DAKE glare / last DAKE1 for re-send on tie-break.
self._last_dake1 = {}
# Terminal-UI state (attached lazily in _start_tui).
self.panel_manager = None
self._screen = None
self._tui_enabled = False
self._tui_last_panel = None
self._tui_autofocused = False
self._tui_jid_by_label = {}
self._tui_label_by_jid = {}
self._own_bare = jid.split("/", 1)[0] if jid else ""
self._probe = False
self._prompt_refresh_cb = None
self.nick = jid.split("@", 1)[0] if jid else "me"
self._keepalive_task = None
# OTR engine.
tracer = OTRTracer(enabled=True) if OTRTracer else None
if tracer is not None and hasattr(tracer, "set_emit_callback"):
def _trace_emit(line, *_a, **_k):
try:
print(f"[otr-trace] {line}")
except Exception:
pass
tracer.set_emit_callback(_trace_emit)
cfg = OTRConfig(test_mode=True)
self.otr = EnhancedSessionManager(config=cfg, tracer=tracer)
# Dedicated single-thread executor for OTR/SMP crypto. SMP runs
# multi-minute 3072-bit DH computations; a separate pool keeps the
# event loop free so keepalive/network stay alive throughout.
self._otr_executor = ThreadPoolExecutor(
max_workers=2, thread_name_prefix="otr-crypto"
)
# Security: never auto-approve subscription requests.
self.auto_authorize = False
self.auto_subscribe = False
# --- Event handlers ---
self.add_event_handler("session_start", self._on_start)
self.add_event_handler("message", self._on_message)
self.add_event_handler("failed_auth", self._on_failed_auth)
self.add_event_handler("message_error", self._on_message_error)
self.add_event_handler("disconnected", self._on_disconnected)
self.add_event_handler("connection_failed", self._on_connection_failed)
self.add_event_handler("stream_error", self._on_stream_error)
self.add_event_handler("presence_subscribe", self._on_subscribe)
self.add_event_handler("presence_subscribed", self._on_subscribed)
self.add_event_handler("presence_available", self._on_presence_available)
self.add_event_handler("presence_unavailable", self._on_presence_unavailable)
self.add_event_handler("receipt_received", self._on_delivery_receipt)
# --- XEP plugins ---
# XEP-0030: Service discovery (required base for many XEPs).
self.register_plugin("xep_0030")
# XEP-0085: Chat state notifications.
self.register_plugin("xep_0085")
# XEP-0115: Entity capabilities (efficient feature advertisement).
self.register_plugin("xep_0115")
# XEP-0184: Message delivery receipts (auto=True: request+send).
self.register_plugin("xep_0184", {"auto": True})
# XEP-0198: Stream management (stanza acks + resumption).
# Degrades gracefully if the server does not advertise SM support.
try:
self.register_plugin("xep_0198", {"max_misses": 3})
except Exception:
pass
# XEP-0199: XMPP Ping (available for /ping command).
self.register_plugin("xep_0199")
# -------------------------------------------------------------------------
# Lifecycle
# -------------------------------------------------------------------------
async def _on_start(self, event):
self.send_presence()
try:
await self.get_roster()
except (IqError, IqTimeout):
pass
print(f"\n[connected] {self.boundjid.full}")
print(f"[version] OTRv4+ XMPP {XMPP_VERSION}")
if self.peer:
self.send_presence_subscription(pto=self.peer)
print(f"[subscribe] requested presence from {self.peer}")
print(
"[ready] /otr to start encryption. After DAKE you'll be asked to "
"trust the fingerprint, then to set the SMP passphrase.\n"
"[ready] Type /help for the full command list.\n"
)
# Reset reconnect backoff on successful connection.
self._reconnect_delay = _RECONNECT_BASE
# Whitespace keepalive to maintain I2P SAM streams during long SMP
# computations when no application data flows.
self._keepalive_task = asyncio.ensure_future(self._keepalive_loop())
async def _keepalive_loop(self):
"""Send a whitespace ping every 8s so idle I2P tunnels stay alive.
The tick counter confirms the event loop is not frozen during SMP."""
n = 0
try:
while self.is_connected():
await asyncio.sleep(8)
n += 1
try:
self.send_raw(" ")
print(f"[keepalive] tick {n} (loop alive)")
except Exception:
break
except asyncio.CancelledError:
pass
def _on_failed_auth(self, event):
print("\n[auth failed] check JID and password.", file=sys.stderr)
# Don't retry on bad credentials; reconnect would loop forever.
self._shutting_down = True
def _on_disconnected(self, event):
print("\n[disconnected]")
if self._keepalive_task:
self._keepalive_task.cancel()
if not self._shutting_down and self._sam_params is not None:
try:
loop = asyncio.get_event_loop()
self._reconnect_task = loop.create_task(self._reconnect())
except Exception as e:
print(f"[reconnect] could not schedule: {e}")
def _on_connection_failed(self, event):
reason = str(event) if event else "unknown"
print(f"[connection failed] {_sanitise(reason, 256)}")
if not self._shutting_down and self._sam_params is not None:
try:
loop = asyncio.get_event_loop()
loop.create_task(self._reconnect())
except Exception:
pass
def _on_stream_error(self, error):
condition = getattr(error, "condition", None) or str(error)
print(f"[stream error] {_sanitise(str(condition), 256)}")
async def _reconnect(self):
"""Exponential-backoff reconnect. Re-establishes the I2P SAM tunnel
before reconnecting slixmpp when running over I2P."""
while not self._shutting_down:
delay = self._reconnect_delay
print(f"[reconnect] waiting {delay}s before reconnecting...")
await asyncio.sleep(delay)
if self._shutting_down:
return
print("[reconnect] attempting reconnection...")
try:
if self._is_i2p and self._sam_params:
p = self._sam_params
try:
host, port = await start_i2p_sam_forwarder(
p["server_b32"],
p["dest_port"],
sam_host=p["sam_host"],
sam_port=p["sam_port"],
)
except Exception as e:
print(f"[reconnect] SAM bridge failed: {e}")
self._reconnect_delay = min(
self._reconnect_delay * 2, _RECONNECT_MAX
)
continue
self.connect(host, port)
else:
self.connect()
print("[reconnect] reconnected.")
return # _on_start resets _reconnect_delay on success
except Exception as e:
print(f"[reconnect] failed: {e}")
self._reconnect_delay = min(
self._reconnect_delay * 2, _RECONNECT_MAX
)
def _on_message_error(self, msg):
peer = msg["from"].bare
text = msg["error"]["text"] or msg["error"]["condition"]
print(f"\n[delivery rejected] to {_sanitise(peer, 128)}: {_sanitise(text)}")
if msg["error"]["condition"] == "forbidden":
print(
" -> not mutually subscribed; both accounts must accept each "
"other as contacts.\n"
)
def _on_delivery_receipt(self, receipt):
"""XEP-0184: fired when a peer acknowledges delivery of our message."""
try:
peer = receipt["from"].bare
msg_id = receipt.get("id", "?")
print(f"[receipt] delivered to {_sanitise(peer, 128)} (id {msg_id})")
except Exception:
pass
# -------------------------------------------------------------------------
# Presence handling
# -------------------------------------------------------------------------
def _on_subscribe(self, presence):
"""Gate subscription requests; never auto-approve."""
peer = presence["from"].bare
self._pending_subscriptions[peer] = presence
print(f"[sub] {_sanitise(peer, 128)} requests subscription.")
print(f"[sub] Type /accept {peer} to approve or /deny {peer} to reject.")
def _on_subscribed(self, presence):
peer = presence["from"].bare
print(f"[sub] {_sanitise(peer, 128)} approved our subscription")
self.send_presence(pto=peer)
def _on_presence_available(self, presence):
peer = presence["from"].bare
if peer == self._own_bare:
return
show = presence["show"] or "available"
status = presence["status"] or ""
status_s = f" ({_sanitise(status, 64)})" if status else ""
print(f"[presence] {_sanitise(peer, 128)} is {show}{status_s}")
def _on_presence_unavailable(self, presence):
peer = presence["from"].bare
if peer == self._own_bare:
return
print(f"[presence] {_sanitise(peer, 128)} went offline")
# -------------------------------------------------------------------------
# Rate limiting
# -------------------------------------------------------------------------
def _check_rate_limit(self, peer: str) -> bool:
"""Return True if the message should be processed; False if throttled."""
now = time.monotonic()
if peer not in self._rate_limit:
self._rate_limit[peer] = collections.deque()
dq = self._rate_limit[peer]
while dq and dq[0] < now - _RATE_WINDOW:
dq.popleft()
if len(dq) >= _RATE_MAX:
return False
dq.append(now)
return True
# -------------------------------------------------------------------------
# Inbound message routing
# -------------------------------------------------------------------------
def _on_message(self, msg):
if msg["type"] not in ("chat", "normal"):
return
peer = msg["from"].bare
body = msg["body"]
if not body:
return
# Session-local block list check.
if peer in self._blocked:
return
# Rate limiting check.
if not self._check_rate_limit(peer):
print(f"[rate-limit] dropping message from {_sanitise(peer, 128)}")
return
# Inbound fragment reassembly.
if body.startswith("?OTRv4F|"):
full = self._reassemble_fragment(peer, body)
if full is None:
return
body = full
if body.startswith(OTR_PREFIX):
# OTR processing (especially SMP) can run multi-minute 3072-bit DH
# computations that BLOCK. Offload to a thread to keep the asyncio
# event loop free so keepalive and network stay responsive.
asyncio.ensure_future(self._handle_otr_in_async(peer, body))
else:
print(f"[plain] <{_sanitise(peer, 128)}> {_sanitise(body)}")
async def _handle_otr_in_async(self, peer, body):
stage_in = self._otr_stage(body)
if stage_in:
print(f"[otr-recv] <- {stage_in} from {peer}")
if self._probe:
try:
keys = sorted(self.otr.sessions.keys())
present = peer in self.otr.sessions
print(
f"[otr-probe] inbound {stage_in}: lookup={peer!r} "
f"present={present} stored_keys={keys}"
)
if not present and keys:
for k in keys:
print(
f"[otr-probe] key mismatch? stored={k!r} "
f"== lookup={peer!r} -> {k == peer} "
f"(len {len(k)} vs {len(peer)})"
)
except Exception as e:
print(f"[otr-probe] inbound probe error: {e}")
# --- DAKE glare resolution ---
# Over slow I2P both sides may send DAKE1 before either receives the
# other's. Tie-break by bare JID: lower JID keeps initiator role;
# higher JID yields and answers as responder. Both sides run identical
# code so exactly one yields.
if stage_in == "DAKE1":
sess = self.otr.get_session(peer)
st = getattr(getattr(sess, "session_state", None), "name", "")
is_init = bool(getattr(sess, "is_initiator", False))
if sess is not None and st == "DAKE_IN_PROGRESS" and is_init:
if self._own_bare < peer:
print(
f"[otr] simultaneous start with {peer}: keeping "
f"initiator role; re-sending our DAKE1"
)
d1 = self._last_dake1.get(peer)
if d1:
self.send_otr_fragmented(peer, d1)
return
print(
f"[otr] simultaneous start with {peer}: yielding initiator "
f"role, answering as responder"
)
try:
self.otr.end_session(peer)
self._last_dake1.pop(peer, None)
self._encrypted.discard(peer)
except Exception as e:
print(f"[otr] glare teardown error: {e}")
heavy = (stage_in or "").startswith("DATA")
if heavy:
import time as _t
t0 = _t.time()
print(
f"[otr-crypto] processing DATA from {peer} "
f"(SMP DH may take minutes; loop stays alive)..."
)
loop = asyncio.get_event_loop()
try:
out = await loop.run_in_executor(
self._otr_executor, self.otr.handle_incoming_message, peer, body
)
except Exception as e:
print(f"[otr error] from {peer}: {e}")
return
if heavy:
import time as _t
print(f"[otr-crypto] done processing DATA from {peer} ({_t.time() - t0:.1f}s).")
self._check_dake_complete(peer)
if out:
out_b = out.encode("utf-8") if isinstance(out, str) else out
if out_b.startswith(OTR_PREFIX_B):
stage_out = self._otr_stage(out_b.decode("utf-8", errors="replace"))
if stage_out:
print(f"[otr-send] -> {stage_out} to {peer}")
self.send_otr_fragmented(peer, out_b.decode("utf-8", errors="replace"))
else:
text = out_b.decode("utf-8", errors="replace")
print(f"[otr] <{_sanitise(peer, 128)}> {_sanitise(text)}")
self._report_smp(peer)
self._check_dake_complete(peer)
@staticmethod
def _otr_stage(frame):
"""Identify the OTRv4 message stage from a '?OTRv4 <base64>' frame.
Best-effort; used for progress display only."""
import base64 as _b64
try:
if not frame.startswith(OTR_PREFIX):
return None
payload = frame[len(OTR_PREFIX):].strip()
if payload.endswith("."):
payload = payload[:-1]
try:
decoded = _b64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4))
except Exception:
std = payload.replace("-", "+").replace("_", "/")
decoded = _b64.b64decode(std + "=" * (-len(std) % 4))
if len(decoded) < 1:
return None
if (
len(decoded) >= 3
and decoded[0] == 0x00
and decoded[1] == 0x04
and decoded[2] == 0x03
):
return "DATA (may carry SMP)"
mtype = decoded[0]
names = {0x35: "DAKE1", 0x36: "DAKE2", 0x37: "DAKE3", 0x03: "DATA"}
return names.get(mtype, f"type 0x{mtype:02x}")
except Exception:
return None
def _handle_otr_in(self, peer, body):
"""Sync fallback (retained for compatibility; async path is preferred)."""
try:
out = self.otr.handle_incoming_message(peer, body)
except Exception as e:
print(f"[otr error] from {peer}: {e}")
return
self._check_dake_complete(peer)
if out:
out_b = out.encode("utf-8") if isinstance(out, str) else out
if out_b.startswith(OTR_PREFIX_B):
self.send_otr_fragmented(peer, out_b.decode("utf-8", errors="replace"))
else:
text = out_b.decode("utf-8", errors="replace")
print(f"[otr] <{_sanitise(peer, 128)}> {_sanitise(text)}")
self._report_smp(peer)
self._check_dake_complete(peer)
# -------------------------------------------------------------------------
# DAKE completion -> trust prompt
# -------------------------------------------------------------------------
def _check_dake_complete(self, peer):
"""When a peer's session first becomes encrypted, show fingerprints
and prompt for trust - identical to the IRC client."""
try:
if not self.otr.has_encrypted_session(peer):
return
except Exception:
return
if peer in self._encrypted:
return
self._encrypted.add(peer)
local_fp = self._local_fp()
remote_fp = self._remote_fp(peer)
print("\n" + "-" * 60)
print(
f"[secure] OTR session with {peer} is ENCRYPTED "
"(X448 + ML-KEM-1024 + ML-DSA-87)."
)
print(f" Your fingerprint : {_fmt_fp(local_fp)}")
print(f" Their fingerprint : {_fmt_fp(remote_fp)}")
print("-" * 60)
already = False
try:
already = self.otr.is_peer_trusted(peer)
except Exception:
already = False
if already:
print("[trust] Fingerprint already trusted - VERIFIED.")
self._prompt_smp_secret(peer)
else:
print("[trust] Trust this fingerprint? Type y or n :")
self._pending[peer] = "trust"
def _handle_trust_answer(self, peer, answer):
ans = answer.strip().lower()
if ans in ("y", "yes"):
ok = False
try:
remote_fp = self._remote_fp(peer)
ok = self.otr.trust_fingerprint(peer, remote_fp)
except Exception as e:
print(f"[trust] error saving trust: {e}")
if ok:
print("[trust] Fingerprint TRUSTED - identity pinned (VERIFIED).")
else:
print("[trust] Could not store trust, continuing encrypted-only.")
else:
print("[trust] Fingerprint NOT trusted - encrypted only.")
self._pending[peer] = None
self._prompt_smp_secret(peer)
# -------------------------------------------------------------------------
# SMP passphrase prompt
# -------------------------------------------------------------------------
def _prompt_smp_secret(self, peer):
print("-" * 60)
print(
"[smp] SOCIALIST MILLIONAIRE PROTOCOL setup "
"(hybrid PQC: ML-KEM-1024 + ML-DSA-87 + ZKP)."
)
print(
f"[smp] Passphrase: {SMP_MIN_LEN}-{SMP_MAX_LEN} chars. "
"Both sides must use the SAME secret."
)
print("[smp] After both have stored it, run /smp start (either side).")
print("[smp] Press Enter or type skip to skip for now.")
self._pending[peer] = "smp_secret"
def _handle_smp_secret_answer(self, peer, secret):
self._pending[peer] = None
if not secret or secret.strip().lower() == "skip":
print("[smp] skipped - you can set it later with /smp-secret <secret>.")
return
secret = secret.strip()
err = self._validate_smp_secret(secret)
if err:
print(f"[smp] {err}")
return
try:
ok = self.otr.set_smp_secret(peer, secret)
except Exception as e:
print(f"[smp] error storing passphrase: {e}")
return
if ok:
print("[smp] passphrase stored for auto-respond.")
print("[smp] When BOTH sides have stored it, run /smp start to verify.")
else:
print("[smp] could not store passphrase.")
@staticmethod
def _validate_smp_secret(secret: str):
"""Return an error string if the SMP secret fails validation, else None."""
if len(secret) < SMP_MIN_LEN:
return f"secret too short (minimum {SMP_MIN_LEN} characters)"
if len(secret) > SMP_MAX_LEN:
return f"secret too long (maximum {SMP_MAX_LEN} characters)"
return None
# -------------------------------------------------------------------------
# SMP result reporting
# -------------------------------------------------------------------------
def _report_smp(self, peer):
try:
session = self.otr.get_session(peer)
if not session:
return
state = getattr(session, "smp_state", None)
if state is None:
return
name = getattr(state, "name", str(state))
key = (peer, name)
if name == "SUCCEEDED" and key not in self._smp_reported:
self._smp_reported.add(key)
print(
f"\n[smp] *** IDENTITY VERIFIED with {peer} - "
"shared secret matched (SMP complete). ***\n"
)
elif name == "FAILED" and key not in self._smp_reported:
self._smp_reported.add(key)
print(
f"\n[smp] *** SMP FAILED with {peer} - secrets did NOT "
"match (or protocol error). Possible MITM. ***\n"
)
except Exception:
pass
# -------------------------------------------------------------------------
# Fingerprint helpers
# -------------------------------------------------------------------------
def _local_fp(self):
try:
cp = getattr(self.otr, "client_profile", None)
if cp and hasattr(cp, "get_fingerprint"):
return cp.get_fingerprint() or "unavailable"
except Exception:
pass
return "unavailable"
def _remote_fp(self, peer):
try:
if hasattr(self.otr, "get_peer_fingerprint"):
fp = self.otr.get_peer_fingerprint(peer)
if fp:
return fp
sess = self.otr.get_session(peer)
if sess and hasattr(sess, "get_fingerprint"):
fp = sess.get_fingerprint()
if fp:
return fp
except Exception:
pass
return "unavailable"
# -------------------------------------------------------------------------
# Outbound fragmentation
# -------------------------------------------------------------------------
def send_otr_fragmented(self, peer, payload):
"""Send an OTR message, fragmenting if over the I2P cliff (~8KB).
Fragment wire format (one <body> per fragment):
?OTRv4F|<msg_id>|<n>|<total>|<chunk>
Small messages are sent whole as a normal ?OTRv4 frame. The monotonic
msg_id avoids collision when two large in-flight DATA frames have
near-identical headers (version + instance tags + ratchet header).
"""
MAX_FRAGMENT = 6000 # bytes per fragment (safely under I2P cliff)
if len(payload) <= MAX_FRAGMENT:
self.send_message(mto=peer, mbody=payload, mtype="chat")
print(f"[otr-send] 1 frame ({len(payload)} bytes) -> {peer}")
return
chunks = [
payload[i : i + MAX_FRAGMENT]
for i in range(0, len(payload), MAX_FRAGMENT)