distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Peer profiles, selection, and banning.
2
3Ported from net.i2p.router.peermanager.PeerProfile.
4"""
5
6import time
7
8
9class PeerProfile:
10 """Full peer profile with reputation metrics.
11
12 Ported from net.i2p.router.peermanager.PeerProfile.
13 """
14
15 def __init__(self, router_hash: bytes):
16 self.router_hash = router_hash
17
18 # Tunnel building
19 self.tunnel_builds_succeeded = 0
20 self.tunnel_builds_failed = 0
21 self.tunnel_builds_rejected = 0
22
23 # Communication
24 self.send_success_count = 0
25 self.send_failure_count = 0
26 self.db_store_success_count = 0
27 self.db_store_failure_count = 0
28 self.db_lookup_success_count = 0
29 self.db_lookup_failure_count = 0
30
31 # Timing
32 self.last_heard_from: float = 0.0
33 self.first_heard_about: float = 0.0
34 self.last_send_success: float = 0.0
35 self.last_send_failure: float = 0.0
36
37 # Computed scores (updated by calculators)
38 self.capacity: float = 0.0
39 self.speed: float = 0.0
40 self.integration: float = 0.0
41
42 # Classification
43 self.is_expanding: bool = False
44 self.is_active: bool = True
45 self.is_banned: bool = False
46 self.ban_expiry: float = 0.0
47
48 # Latency tracking
49 self._latency_samples: list[float] = []
50 self._max_latency_samples = 50
51
52 # Legacy compat
53 self._successes = 0
54 self._failures = 0
55
56 # --- Tunnel build recording ---
57
58 def record_tunnel_build_success(self) -> None:
59 self.tunnel_builds_succeeded += 1
60
61 def record_tunnel_build_failure(self) -> None:
62 self.tunnel_builds_failed += 1
63
64 def record_tunnel_build_rejection(self) -> None:
65 self.tunnel_builds_rejected += 1
66
67 # --- Communication recording ---
68
69 def record_send_success(self) -> None:
70 self.send_success_count += 1
71 self.last_send_success = time.time()
72
73 def record_send_failure(self) -> None:
74 self.send_failure_count += 1
75 self.last_send_failure = time.time()
76
77 def record_db_store_success(self) -> None:
78 self.db_store_success_count += 1
79
80 def record_db_store_failure(self) -> None:
81 self.db_store_failure_count += 1
82
83 def record_db_lookup_success(self) -> None:
84 self.db_lookup_success_count += 1
85
86 def record_db_lookup_failure(self) -> None:
87 self.db_lookup_failure_count += 1
88
89 # --- Latency ---
90
91 def record_latency(self, latency_ms: float) -> None:
92 self._latency_samples.append(latency_ms)
93 if len(self._latency_samples) > self._max_latency_samples:
94 self._latency_samples = self._latency_samples[-self._max_latency_samples:]
95
96 # --- Activity ---
97
98 def heard_from(self) -> None:
99 now = time.time()
100 self.last_heard_from = now
101 if self.first_heard_about == 0.0:
102 self.first_heard_about = now
103
104 # --- Computed properties ---
105
106 @property
107 def tunnel_success_rate(self) -> float:
108 total = self.tunnel_builds_succeeded + self.tunnel_builds_failed
109 if total == 0:
110 return 0.0
111 return self.tunnel_builds_succeeded / total
112
113 @property
114 def send_success_rate(self) -> float:
115 total = self.send_success_count + self.send_failure_count
116 if total == 0:
117 return 0.0
118 return self.send_success_count / total
119
120 @property
121 def db_success_rate(self) -> float:
122 total = (self.db_store_success_count + self.db_store_failure_count +
123 self.db_lookup_success_count + self.db_lookup_failure_count)
124 if total == 0:
125 return 0.0
126 successes = self.db_store_success_count + self.db_lookup_success_count
127 return successes / total
128
129 @property
130 def average_latency(self) -> float:
131 if not self._latency_samples:
132 return 0.0
133 return sum(self._latency_samples) / len(self._latency_samples)
134
135 @property
136 def is_established(self) -> bool:
137 """Has enough history to be classified."""
138 total_tunnel = self.tunnel_builds_succeeded + self.tunnel_builds_failed
139 total_send = self.send_success_count + self.send_failure_count
140 return total_tunnel >= 5 and total_send >= 5
141
142 # --- Ban management ---
143
144 def ban(self, duration_seconds: float) -> None:
145 self.is_banned = True
146 self.ban_expiry = time.time() + duration_seconds
147
148 def unban(self) -> None:
149 self.is_banned = False
150 self.ban_expiry = 0.0
151
152 @property
153 def is_currently_banned(self) -> bool:
154 if not self.is_banned:
155 return False
156 if time.time() >= self.ban_expiry:
157 self.is_banned = False
158 self.ban_expiry = 0.0
159 return False
160 return True
161
162 # --- Serialization ---
163
164 def to_dict(self) -> dict:
165 return {
166 "router_hash": self.router_hash.hex(),
167 "tunnel_builds_succeeded": self.tunnel_builds_succeeded,
168 "tunnel_builds_failed": self.tunnel_builds_failed,
169 "tunnel_builds_rejected": self.tunnel_builds_rejected,
170 "send_success_count": self.send_success_count,
171 "send_failure_count": self.send_failure_count,
172 "db_store_success_count": self.db_store_success_count,
173 "db_store_failure_count": self.db_store_failure_count,
174 "db_lookup_success_count": self.db_lookup_success_count,
175 "db_lookup_failure_count": self.db_lookup_failure_count,
176 "last_heard_from": self.last_heard_from,
177 "first_heard_about": self.first_heard_about,
178 "last_send_success": self.last_send_success,
179 "last_send_failure": self.last_send_failure,
180 "capacity": self.capacity,
181 "speed": self.speed,
182 "integration": self.integration,
183 "is_expanding": self.is_expanding,
184 "is_active": self.is_active,
185 "is_banned": self.is_banned,
186 "ban_expiry": self.ban_expiry,
187 "latency_samples": list(self._latency_samples),
188 }
189
190 @classmethod
191 def from_dict(cls, data: dict) -> "PeerProfile":
192 p = cls(bytes.fromhex(data["router_hash"]))
193 p.tunnel_builds_succeeded = data["tunnel_builds_succeeded"]
194 p.tunnel_builds_failed = data["tunnel_builds_failed"]
195 p.tunnel_builds_rejected = data["tunnel_builds_rejected"]
196 p.send_success_count = data["send_success_count"]
197 p.send_failure_count = data["send_failure_count"]
198 p.db_store_success_count = data["db_store_success_count"]
199 p.db_store_failure_count = data["db_store_failure_count"]
200 p.db_lookup_success_count = data["db_lookup_success_count"]
201 p.db_lookup_failure_count = data["db_lookup_failure_count"]
202 p.last_heard_from = data["last_heard_from"]
203 p.first_heard_about = data["first_heard_about"]
204 p.last_send_success = data["last_send_success"]
205 p.last_send_failure = data["last_send_failure"]
206 p.capacity = data["capacity"]
207 p.speed = data["speed"]
208 p.integration = data["integration"]
209 p.is_expanding = data["is_expanding"]
210 p.is_active = data["is_active"]
211 p.is_banned = data["is_banned"]
212 p.ban_expiry = data["ban_expiry"]
213 p._latency_samples = list(data.get("latency_samples", []))
214 return p
215
216 # --- Legacy compatibility ---
217
218 def record_success(self):
219 self._successes += 1
220 self.capacity = self._successes / (self._successes + self._failures + 1)
221
222 def record_failure(self):
223 self._failures += 1
224 self.capacity = self._successes / (self._successes + self._failures + 1)
225
226 def tunnel_build_success(self):
227 self.tunnel_builds_succeeded += 1
228
229 def tunnel_build_failure(self):
230 self.tunnel_builds_failed += 1
231
232 def update_last_heard(self, timestamp_ms: int):
233 self.last_heard_from = timestamp_ms
234
235 def decay(self, factor: float = 0.95):
236 self.capacity *= factor
237 self._successes = int(self._successes * factor)
238 self._failures = int(self._failures * factor)
239
240 def score(self) -> float:
241 total = self.tunnel_builds_succeeded + self.tunnel_builds_failed
242 tsr = self.tunnel_builds_succeeded / total if total > 0 else 0.0
243 return self.capacity + tsr * 0.5
244
245
246class PeerSelector:
247 """Select peers by profile score."""
248
249 def __init__(self, profiles: list[PeerProfile]):
250 self._profiles = profiles
251
252 def select_best(self, n: int, exclude: set[bytes] | None = None) -> list[PeerProfile]:
253 exclude = exclude or set()
254 candidates = [p for p in self._profiles if p.router_hash not in exclude]
255 candidates.sort(key=lambda p: p.score(), reverse=True)
256 return candidates[:n]
257
258
259class BanManager:
260 """Temporary ban management for misbehaving peers."""
261
262 def __init__(self):
263 self._bans: dict[bytes, dict] = {}
264
265 def ban(self, router_hash: bytes, reason: str, duration_ms: int,
266 now_ms: int | None = None):
267 if now_ms is None:
268 now_ms = int(time.time() * 1000)
269 self._bans[router_hash] = {
270 "reason": reason,
271 "expires": now_ms + duration_ms,
272 }
273
274 def unban(self, router_hash: bytes):
275 self._bans.pop(router_hash, None)
276
277 def is_banned(self, router_hash: bytes, now_ms: int | None = None) -> bool:
278 info = self._bans.get(router_hash)
279 if info is None:
280 return False
281 if now_ms is None:
282 now_ms = int(time.time() * 1000)
283 if now_ms >= info["expires"]:
284 del self._bans[router_hash]
285 return False
286 return True
287
288 def ban_count(self) -> int:
289 return len(self._bans)
290
291 def cleanup(self, now_ms: int | None = None):
292 if now_ms is None:
293 now_ms = int(time.time() * 1000)
294 expired = [h for h, info in self._bans.items() if now_ms >= info["expires"]]
295 for h in expired:
296 del self._bans[h]