2424from sqlmodel .ext .asyncio .session import AsyncSession
2525
2626from app .config import settings
27- from app .core .config_gen import _build_outbound
27+ from app .core .config_gen import _build_outbound , SPEED_PROBE_PORT
2828from app .core .healthcheck import HealthChecker
2929from app .database import get_async_engine
30- from app .models import Node
30+ from app .models import Node , Settings as DBSettings
3131
3232logger = logging .getLogger (__name__ )
3333
7373_REACH_TIMEOUT_S = 6.0
7474
7575
76+ async def _is_active_node (node_id : int ) -> bool :
77+ """Is `node_id` the currently active node? Never raises — a lookup hiccup
78+ just falls back to the temp-xray path."""
79+ try :
80+ async with AsyncSession (get_async_engine ()) as s :
81+ row = (await s .exec (
82+ select (DBSettings ).where (DBSettings .key == "active_node_id" )
83+ )).first ()
84+ return bool (row and row .value and int (row .value ) == node_id )
85+ except (TypeError , ValueError ):
86+ return False
87+ except Exception as exc : # noqa: BLE001 — never block a speed test on this
88+ logger .debug ("active-node lookup failed: %s" , exc )
89+ return False
90+
91+
92+ async def _live_socks_ready (node_id : int ) -> bool :
93+ """True when `node_id` is active AND the main xray's speed-probe inbound is
94+ listening — i.e. we can measure it through the LIVE tunnel instead of a
95+ second temp xray (which, for WireGuard, would fight the live session)."""
96+ return await _is_active_node (node_id ) and await _port_open ("127.0.0.1" , SPEED_PROBE_PORT )
97+
98+
99+ async def _gate_and_measure (node : Node , socks_port : int ) -> Dict :
100+ """Reachability gate + download measurement over an already-open SOCKS
101+ port — a throwaway xray OR the live speed-probe inbound. Shared so both
102+ paths measure identically."""
103+ reachable , latency_ms , detail = await _probe_reachable (socks_port )
104+ if not reachable :
105+ return _result (node , error = f"unreachable: { detail } ,
106+ reachable = False , latency_ms = latency_ms )
107+ last_error = "no targets reachable"
108+ for label , url in _STREAM_TARGETS :
109+ try :
110+ measured = await _measure_download (socks_port , label , url )
111+ except Exception as exc : # noqa: BLE001 — try the next target
112+ last_error = f"{ label } { str (exc )[:120 ]}
113+ continue
114+ if measured is not None :
115+ avg , mx = measured
116+ return _result (node , download_mbps = avg , max_mbps = mx ,
117+ reachable = True , latency_ms = latency_ms )
118+ return _result (node , error = last_error , reachable = True , latency_ms = latency_ms )
119+
120+
76121async def speedtest_node (node : Node ) -> Dict :
77122 """Measure download speed through a freshly-spawned xray for `node`.
78123
@@ -92,6 +137,15 @@ async def speedtest_node(node: Node) -> Dict:
92137 proc : Optional [asyncio .subprocess .Process ] = None
93138 tmp_path : Optional [str ] = None
94139 try :
140+ # Active node: measure through the LIVE tunnel (the main xray's
141+ # speed-probe inbound → active outbound) instead of a throwaway xray.
142+ # A second instance re-opens the node's outbound; for WireGuard that's
143+ # a second session with the same peer key, which the server can't hold
144+ # twice — the temp test and the live tunnel fight, the reachability
145+ # gate flaps to "unreachable", and the live tunnel is briefly disrupted.
146+ if await _live_socks_ready (node .id ):
147+ return await _gate_and_measure (node , SPEED_PROBE_PORT )
148+
95149 # Resolve the full chain: [entry_parent, ..., node]. Only the entry
96150 # parent's address needs to be reachable directly; deeper hops tunnel.
97151 try :
@@ -121,26 +175,7 @@ async def speedtest_node(node: Node) -> Dict:
121175 if proc is None or start_err :
122176 return _result (node , error = start_err or "Failed to start temp xray" )
123177
124- # Reachability gate — skip the speed fallbacks entirely on a dead node.
125- reachable , latency_ms , detail = await _probe_reachable (socks_port )
126- if not reachable :
127- return _result (node , error = f"unreachable: { detail } ,
128- reachable = False , latency_ms = latency_ms )
129-
130- # Measure like the live test: avg after warm-up + peak steady window.
131- # First target that yields data wins.
132- last_error = "no targets reachable"
133- for label , url in _STREAM_TARGETS :
134- try :
135- measured = await _measure_download (socks_port , label , url )
136- except Exception as exc : # noqa: BLE001 — try the next target
137- last_error = f"{ label } { str (exc )[:120 ]}
138- continue
139- if measured is not None :
140- avg , mx = measured
141- return _result (node , download_mbps = avg , max_mbps = mx ,
142- reachable = True , latency_ms = latency_ms )
143- return _result (node , error = last_error , reachable = True , latency_ms = latency_ms )
178+ return await _gate_and_measure (node , socks_port )
144179
145180 except Exception as exc :
146181 logger .warning ("Speedtest node %d error: %s" , node .id , exc )
@@ -167,33 +202,40 @@ async def speedtest_stream(node: Node):
167202 tmp_path : Optional [str ] = None
168203 try :
169204 yield {"phase" : "start" , "node_id" : node .id }
170- try :
171- chain = await _resolve_chain (node )
172- except Exception as exc : # noqa: BLE001
173- yield {"phase" : "error" , "error" : f"chain resolve: { exc } }
174- return
175205
176- entry = chain [0 ]
177- entry_ip : Optional [str ] = None
178- if entry .protocol != "naive" :
206+ # Active node → measure through the LIVE tunnel (speed-probe inbound),
207+ # never a second temp xray (WireGuard single-session contention — see
208+ # speedtest_node). Otherwise spin the usual throwaway instance.
209+ if await _live_socks_ready (node .id ):
210+ socks_port = SPEED_PROBE_PORT
211+ else :
179212 try :
180- entry_ip = await HealthChecker . _resolve_direct ( entry . address )
213+ chain = await _resolve_chain ( node )
181214 except Exception as exc : # noqa: BLE001
182- yield {"phase" : "error" , "error" : f"dns entry: { exc } }
183- return
184- if node .protocol == "naive" :
185- if not node .internal_port :
186- yield {"phase" : "error" , "error" : "naive sidecar port not allocated" }
187- return
188- if not await _port_open ("127.0.0.1" , int (node .internal_port )):
189- yield {"phase" : "error" , "error" : f"naive sidecar not on :{ node .internal_port } }
215+ yield {"phase" : "error" , "error" : f"chain resolve: { exc } }
190216 return
191217
192- yield {"phase" : "connecting" , "host" : "starting xray" }
193- socks_port , proc , tmp_path , start_err = await _start_temp_xray (node , chain , entry_ip )
194- if proc is None or start_err :
195- yield {"phase" : "error" , "error" : start_err or "failed to start temp xray" }
196- return
218+ entry = chain [0 ]
219+ entry_ip : Optional [str ] = None
220+ if entry .protocol != "naive" :
221+ try :
222+ entry_ip = await HealthChecker ._resolve_direct (entry .address )
223+ except Exception as exc : # noqa: BLE001
224+ yield {"phase" : "error" , "error" : f"dns entry: { exc } }
225+ return
226+ if node .protocol == "naive" :
227+ if not node .internal_port :
228+ yield {"phase" : "error" , "error" : "naive sidecar port not allocated" }
229+ return
230+ if not await _port_open ("127.0.0.1" , int (node .internal_port )):
231+ yield {"phase" : "error" , "error" : f"naive sidecar not on :{ node .internal_port } }
232+ return
233+
234+ yield {"phase" : "connecting" , "host" : "starting xray" }
235+ socks_port , proc , tmp_path , start_err = await _start_temp_xray (node , chain , entry_ip )
236+ if proc is None or start_err :
237+ yield {"phase" : "error" , "error" : start_err or "failed to start temp xray" }
238+ return
197239
198240 # Reachability gate first — if the node can't reach the internet, say
199241 # so and stop instead of grinding every speed fallback to timeout.
@@ -272,6 +314,8 @@ async def sni_scan(domain: str, node: Optional[Node] = None) -> Dict:
272314 probed directly. Returns {ok, tls13, http2, status, via, detail}."""
273315 import re
274316
317+ import ipaddress
318+
275319 domain = (domain or "" ).strip ()
276320 for pfx in ("https://" , "http://" ):
277321 if domain .lower ().startswith (pfx ):
@@ -280,6 +324,16 @@ async def sni_scan(domain: str, node: Optional[Node] = None) -> Dict:
280324 if not domain :
281325 return {"ok" : False , "detail" : "empty domain" , "via" : "direct" }
282326
327+ # A bare IP (or any host whose cert won't validate) still needs probing
328+ # so we can read the cert it presents — that's the "scan the IP and see
329+ # what turns up" case. `-k` below lets the handshake finish on a mismatch
330+ # so curl still prints the server-certificate block.
331+ try :
332+ ipaddress .ip_address (domain )
333+ is_ip = True
334+ except ValueError :
335+ is_ip = False
336+
283337 proc : Optional [asyncio .subprocess .Process ] = None
284338 tmp_path : Optional [str ] = None
285339 socks_port : Optional [int ] = None
@@ -306,6 +360,8 @@ async def sni_scan(domain: str, node: Optional[Node] = None) -> Dict:
306360
307361 cmd = ["curl" , "-sS" , "-v" , "-o" , "/dev/null" , "--max-time" , "10" ,
308362 "--http2" , "-A" , _STREAM_UA ]
363+ if is_ip :
364+ cmd .append ("-k" )
309365 if socks_port :
310366 cmd += ["-x" , f"socks5h://127.0.0.1:{ socks_port } ]
311367 cmd .append (f"https://{ domain } )
@@ -323,6 +379,15 @@ async def sni_scan(domain: str, node: Optional[Node] = None) -> Dict:
323379 status = int (m_status .group (1 )) if m_status else None
324380 reachable = status is not None
325381
382+ # Pull the certificate the endpoint actually presents, so scanning a
383+ # bare IP still surfaces the domain(s) behind it (what the operator
384+ # can use as the REALITY serverName). curl prints "subject: CN=…" and,
385+ # on a name match, the cert's own name on the subjectAltName line.
386+ m_subj = re .search (r"subject:\s*([^\r\n]+)" , txt )
387+ cert_subject = m_subj .group (1 ).strip () if m_subj else None
388+ m_san = re .search (r"subjectAltName:[^\r\n]*?cert's \"([^\"]+)\"" , txt )
389+ cert_name = m_san .group (1 ) if m_san else None
390+
326391 ok = bool (tls13 and http2 and reachable and status < 400 )
327392 if ok :
328393 detail = "good REALITY dest"
@@ -341,6 +406,7 @@ async def sni_scan(domain: str, node: Optional[Node] = None) -> Dict:
341406 return {
342407 "ok" : ok , "tls13" : tls13 , "http2" : http2 ,
343408 "status" : status , "via" : via , "detail" : detail ,
409+ "cert_subject" : cert_subject , "cert_name" : cert_name ,
344410 }
345411 except asyncio .TimeoutError :
346412 return {"ok" : False , "detail" : "timed out" , "via" : via }
0 commit comments