distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""RoutingKeyGenerator — date-based routing key derivation for Kademlia.
2
3Ported from net.i2p.data.RoutingKeyGenerator.
4
5Routing keys rotate daily at midnight UTC. The routing key for a given
6database key is SHA-256(key + YYYY-MM-DD_UTF8).
7"""
8
9from __future__ import annotations
10
11import hashlib
12import time
13from datetime import datetime, timezone
14
15
16class RoutingKeyGenerator:
17 """Generate date-based routing keys for NetDB lookups."""
18
19 def __init__(self) -> None:
20 self._mod_data: bytes = b""
21 self._last_changed: float = 0.0
22 self.generate_date_based_mod_data()
23
24 def get_routing_key(self, orig_key: bytes) -> bytes:
25 """Compute the routing key: SHA-256(orig_key + mod_data)."""
26 return hashlib.sha256(orig_key + self._mod_data).digest()
27
28 def generate_date_based_mod_data(self) -> bool:
29 """Update mod_data if the date has changed.
30
31 Returns True if mod_data was updated.
32 """
33 now = datetime.now(timezone.utc)
34 new_mod = now.strftime("%Y-%m-%d").encode("ascii")
35 if new_mod != self._mod_data:
36 self._mod_data = new_mod
37 self._last_changed = time.monotonic()
38 return True
39 return False
40
41 def get_time_till_midnight_utc(self) -> float:
42 """Seconds until the next UTC midnight."""
43 now = datetime.now(timezone.utc)
44 tomorrow = now.replace(
45 hour=0, minute=0, second=0, microsecond=0
46 )
47 # Move to next day
48 from datetime import timedelta
49 tomorrow += timedelta(days=1)
50 return (tomorrow - now).total_seconds()
51
52 @property
53 def mod_data(self) -> bytes:
54 return self._mod_data
55
56 @property
57 def last_changed(self) -> float:
58 return self._last_changed