Horizontal42/netsleuth
1
0
Fork 0
mirror of https://github.com/Horizontal42/netsleuth.git synced 2026-11-11 01:02:10 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge 6f67630963 into 8c99ac78b7

Browse source
This commit is contained in:
Andrew Pimenov 2026-08-26 17:03:46 +03:00 • committed by GitHub
commit bd1192e0a7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 23 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
src/netsleuth/probes/captive_portal.py
View file

@ -1,5 +1,6 @@
from __future__ import annotations
import asyncio
from urllib.parse import urlsplit
import httpx
@ -32,6 +33,23 @@ def classify_portal_response(
return "suspect", f"unexpected status {status}"
async def _fetch_and_classify(
client: httpx.AsyncClient,
url: str,
expected_status: int,
timeout: float,
) -> PortalCheck:
try:
response = await client.get(url, follow_redirects=False, timeout=timeout)
verdict, evidence = classify_portal_response(
url, response.status_code, response.text, response.headers.get("location"), expected_status
)
return PortalCheck(url=url, status=response.status_code, verdict=verdict, evidence=evidence)
except httpx.HTTPError as exc:
verdict, evidence = "error", str(exc)
return PortalCheck(url=url, status=None, verdict=verdict, evidence=evidence)
async def check_captive_portal(
client: httpx.AsyncClient,
urls: list[str],
@ -39,18 +57,11 @@ async def check_captive_portal(
expected_status: int = 204,
timeout: float,
) -> CaptivePortal:
checks: list[PortalCheck] = []
for url in urls:
try:
response = await client.get(url, follow_redirects=False, timeout=timeout)
verdict, evidence = classify_portal_response(
url, response.status_code, response.text, response.headers.get("location"), expected_status
)
except httpx.HTTPError as exc:
verdict, evidence = "error", str(exc)
checks.append(PortalCheck(url=url, status=None, verdict=verdict, evidence=evidence))
continue
checks.append(PortalCheck(url=url, status=response.status_code, verdict=verdict, evidence=evidence))
coros = [
_fetch_and_classify(client, url, expected_status, timeout)
for url in urls
]
checks = list(await asyncio.gather(*coros))
portal_checks = [c for c in checks if c.verdict == "portal"]
if portal_checks: