From 18f956a9cae79222fd85f558d8d78640526169d5 Mon Sep 17 00:00:00 2001 From: parsa Date: Wed, 15 Jul 2026 14:03:27 -0700 Subject: [PATCH 1/4] refactor(core): derive REALITY scan SNI from the target The scanner no longer takes a separate SNI and no longer pulls the inbound's first serverName. The SNI is derived from the target itself: - hostname target -> the hostname is its own SNI, validated against it - bare IP target -> handshake, discover the domain from the presented certificate (CN or first non-wildcard SAN), then re-validate the cert against that discovered domain; the result reports it as discovered from the IP Removes RealityScanRequest.sni and the serverNames->SNI dialog wiring; adds RealityScanResult.sni_discovered. Multi-target scanning is kept (each target derives its own SNI). Adds sniLabel/sniDiscovered locales. --- app/models/reality_scan.py | 2 +- app/operation/core.py | 2 +- app/utils/reality_scan.py | 123 ++++++++++++------ dashboard/public/statics/locales/en.json | 4 +- dashboard/public/statics/locales/fa.json | 4 +- dashboard/public/statics/locales/ru.json | 4 +- dashboard/public/statics/locales/zh.json | 4 +- .../components/xray/reality-scan-dialog.tsx | 34 ++--- .../components/xray/xray-inbounds-section.tsx | 22 +--- dashboard/src/service/reality-scan.ts | 2 +- tests/test_reality_scan_unit.py | 48 ++++++- 11 files changed, 157 insertions(+), 92 deletions(-) diff --git a/app/models/reality_scan.py b/app/models/reality_scan.py index c9a13a4c1..a1d918a36 100644 --- a/app/models/reality_scan.py +++ b/app/models/reality_scan.py @@ -3,7 +3,6 @@ class RealityScanRequest(BaseModel): target: str = Field(min_length=1, max_length=253, description="host or host:port to probe (port defaults to 443)") - sni: str | None = Field(default=None, max_length=253, description="Override the SNI sent during the handshake") timeout: float | None = Field(default=None, ge=1, le=20, description="Per-probe timeout in seconds (1-20, default 10)") @@ -15,6 +14,7 @@ class RealityScanResult(BaseModel): ip: str | None = None port: int sni: str | None = None + sni_discovered: bool = False feasible: bool tls13: bool diff --git a/app/operation/core.py b/app/operation/core.py index acdcca8fc..58bdbccf9 100644 --- a/app/operation/core.py +++ b/app/operation/core.py @@ -40,7 +40,7 @@ async def _refresh_hosts_from_db(self, db: AsyncSession) -> None: async def scan_reality_target(self, request: RealityScanRequest) -> RealityScanResult: try: - result = await scan_reality_target(target=request.target, sni=request.sni, timeout=request.timeout) + result = await scan_reality_target(target=request.target, timeout=request.timeout) except RealityScanError as e: await self.raise_error(message=str(e), code=400) except Exception as e: diff --git a/app/utils/reality_scan.py b/app/utils/reality_scan.py index dff6efeb5..b7275e020 100644 --- a/app/utils/reality_scan.py +++ b/app/utils/reality_scan.py @@ -56,11 +56,11 @@ def _has_control_chars(value: str) -> bool: return any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in value) -def parse_target(target: str, sni_override: str | None = None) -> tuple[str, int, str | None]: +def parse_target(target: str) -> tuple[str, int, str | None]: if not target or not target.strip(): raise RealityScanError("A target host is required.") - if _has_control_chars(target) or (sni_override and _has_control_chars(sni_override)): - raise RealityScanError("Target and SNI must not contain control characters.") + if _has_control_chars(target): + raise RealityScanError("Target must not contain control characters.") value = target.strip() if "://" in value: @@ -87,12 +87,7 @@ def parse_target(target: str, sni_override: str | None = None) -> tuple[str, int if not host: raise RealityScanError("A target host is required.") - sni: str | None - if sni_override and sni_override.strip(): - sni = sni_override.strip() - else: - sni = None if _is_ip_literal(host) else host - + sni = None if _is_ip_literal(host) else host return host, port, sni @@ -240,6 +235,43 @@ def _parse_certificate(der: bytes | None) -> dict: return out +def _first_usable_name(der: bytes | None) -> str | None: + if not der: + return None + try: + cert = x509.load_der_x509_certificate(der) + except Exception: + return None + cn = _name_common_name(cert.subject) + if cn and not cn.startswith("*."): + return cn.strip() + try: + san = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME) + for name in san.value.get_values_for_type(x509.DNSName): + name = name.strip() + if name and not name.startswith("*."): + return name + except Exception: + pass + return None + + +def _make_verify_ctx() -> ssl.SSLContext: + ctx = ssl.create_default_context() + ctx.minimum_version = ssl.TLSVersion.TLSv1_2 + ctx.set_alpn_protocols(["h2", "http/1.1"]) + return ctx + + +def _make_permissive_ctx() -> ssl.SSLContext: + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.minimum_version = ssl.TLSVersion.TLSv1_2 + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + ctx.set_alpn_protocols(["h2", "http/1.1"]) + return ctx + + def _tls_probe(ip: str, port: int, sni: str | None, timeout: float) -> dict: result: dict = { "tls13": False, @@ -253,44 +285,48 @@ def _tls_probe(ip: str, port: int, sni: str | None, timeout: float) -> dict: "server_names": [], "latency_ms": None, "reason": None, + "sni": sni, + "sni_discovered": False, } - verify_ctx = ssl.create_default_context() - verify_ctx.minimum_version = ssl.TLSVersion.TLSv1_2 - verify_ctx.set_alpn_protocols(["h2", "http/1.1"]) - if sni is None: - verify_ctx.check_hostname = False - verify_ctx.verify_mode = ssl.CERT_NONE - - permissive_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - permissive_ctx.minimum_version = ssl.TLSVersion.TLSv1_2 - permissive_ctx.check_hostname = False - permissive_ctx.verify_mode = ssl.CERT_NONE - permissive_ctx.set_alpn_protocols(["h2", "http/1.1"]) - - der: bytes | None = None version: str | None = None alpn: str | None = None + der: bytes | None = None - def _handshake(ctx: ssl.SSLContext) -> tuple[str | None, str | None, bytes | None, float]: + def _handshake(ctx: ssl.SSLContext, server_hostname: str | None) -> tuple[str | None, str | None, bytes | None, float]: started = time.monotonic() with socket.create_connection((ip, port), timeout=timeout) as sock: - with ctx.wrap_socket(sock, server_hostname=sni) as tls: + with ctx.wrap_socket(sock, server_hostname=server_hostname) as tls: latency = (time.monotonic() - started) * 1000.0 return tls.version(), tls.selected_alpn_protocol(), tls.getpeercert(binary_form=True), latency try: - version, alpn, der, latency = _handshake(verify_ctx) - result["cert_valid"] = sni is not None - result["latency_ms"] = round(latency) - except ssl.SSLCertVerificationError as exc: - result["reason"] = f"Certificate did not validate: {getattr(exc, 'verify_message', None) or exc}" - try: - version, alpn, der, latency = _handshake(permissive_ctx) + if sni is not None: + try: + version, alpn, der, latency = _handshake(_make_verify_ctx(), sni) + result["cert_valid"] = True + result["latency_ms"] = round(latency) + except ssl.SSLCertVerificationError as exc: + result["reason"] = f"Certificate did not validate: {getattr(exc, 'verify_message', None) or exc}" + version, alpn, der, latency = _handshake(_make_permissive_ctx(), sni) + result["latency_ms"] = round(latency) + else: + version, alpn, der, latency = _handshake(_make_permissive_ctx(), None) result["latency_ms"] = round(latency) - except Exception as exc2: - result["reason"] = f"TLS handshake failed: {exc2}" - return result + discovered = _first_usable_name(der) + if not discovered: + result["reason"] = "No usable domain name found in the certificate." + else: + result["sni"] = discovered + result["sni_discovered"] = True + try: + version, alpn, der, latency = _handshake(_make_verify_ctx(), discovered) + result["cert_valid"] = True + result["latency_ms"] = round(latency) + except ssl.SSLCertVerificationError as exc: + result["reason"] = f"Certificate did not validate: {getattr(exc, 'verify_message', None) or exc}" + except (ssl.SSLError, socket.timeout, TimeoutError, OSError) as exc: + result["reason"] = f"Certificate re-validation failed: {exc}" except (socket.timeout, TimeoutError): result["reason"] = "Connection timed out." return result @@ -301,6 +337,9 @@ def _handshake(ctx: ssl.SSLContext) -> tuple[str | None, str | None, bytes | Non result["reason"] = f"Connection failed: {exc}" return result + if version is None: + return result + result["tls_version"] = _pretty_tls_version(version) result["tls13"] = version == "TLSv1.3" result["alpn"] = alpn @@ -521,13 +560,15 @@ def _h3_probe(host: str, ip: str, port: int, sni: str | None, timeout: float) -> def _scan_sync(host: str, ip: str, port: int, sni: str | None, timeout: float) -> dict: tls = _tls_probe(ip, port, sni, timeout) + effective_sni = tls["sni"] result: dict = { "target": f"{host}:{port}", "host": host, "ip": ip, "port": port, - "sni": sni, + "sni": effective_sni, + "sni_discovered": tls["sni_discovered"], "feasible": False, "tls13": tls["tls13"], "tls_version": tls["tls_version"], @@ -549,23 +590,23 @@ def _scan_sync(host: str, ip: str, port: int, sni: str | None, timeout: float) - if tls["tls_version"] is None: return result - group = _group_probe(ip, port, sni, timeout) + group = _group_probe(ip, port, effective_sni, timeout) result["x25519"] = group["x25519"] result["post_quantum"] = group["post_quantum"] result["curve"] = group["curve"] - result["h3"] = _h3_probe(host, ip, port, sni, timeout) + result["h3"] = _h3_probe(host, ip, port, effective_sni, timeout) definitely_not_x25519 = group["x25519"] is False and group["post_quantum"] is False and group["curve"] is not None result["feasible"] = bool(result["tls13"] and result["h2"] and result["cert_valid"] and not definitely_not_x25519) return result -async def scan_reality_target(target: str, sni: str | None = None, timeout: float | None = None) -> dict: - host, port, resolved_sni = parse_target(target, sni) +async def scan_reality_target(target: str, timeout: float | None = None) -> dict: + host, port, sni = parse_target(target) clamped = _clamp_timeout(timeout) async with _get_scan_semaphore(): ip = await _resolve_public_ip_async(host, min(clamped, DNS_TIMEOUT)) try: - return await asyncio.wait_for(asyncio.to_thread(_scan_sync, host, ip, port, resolved_sni, clamped), timeout=clamped * 6 + 15) + return await asyncio.wait_for(asyncio.to_thread(_scan_sync, host, ip, port, sni, clamped), timeout=clamped * 6 + 15) except TimeoutError: raise RealityScanError("Scan timed out.") diff --git a/dashboard/public/statics/locales/en.json b/dashboard/public/statics/locales/en.json index 467b415ff..8806cfb85 100644 --- a/dashboard/public/statics/locales/en.json +++ b/dashboard/public/statics/locales/en.json @@ -3071,7 +3071,9 @@ "expires": "Expires", "serverNames": "Certificate server names (valid SNIs)", "unknown": "Unknown", - "other": "Other" + "other": "Other", + "sniLabel": "SNI", + "sniDiscovered": "discovered from IP" } }, "settings.cores.title": "Cores", diff --git a/dashboard/public/statics/locales/fa.json b/dashboard/public/statics/locales/fa.json index 02c1a37bc..efbfbbbab 100644 --- a/dashboard/public/statics/locales/fa.json +++ b/dashboard/public/statics/locales/fa.json @@ -2983,7 +2983,9 @@ "expires": "انقضا", "serverNames": "نام‌های سرور گواهی (SNIهای معتبر)", "unknown": "نامشخص", - "other": "سایر" + "other": "سایر", + "sniLabel": "SNI", + "sniDiscovered": "کشف‌شده از IP" } }, "settings.cores.title": "هسته‌ها", diff --git a/dashboard/public/statics/locales/ru.json b/dashboard/public/statics/locales/ru.json index 9532d9ea3..06de0f1e3 100644 --- a/dashboard/public/statics/locales/ru.json +++ b/dashboard/public/statics/locales/ru.json @@ -2956,7 +2956,9 @@ "expires": "Истекает", "serverNames": "Имена серверов сертификата (действительные SNI)", "unknown": "Неизвестно", - "other": "Другое" + "other": "Другое", + "sniLabel": "SNI", + "sniDiscovered": "обнаружено по IP" } }, "settings.cores.title": "Ядра", diff --git a/dashboard/public/statics/locales/zh.json b/dashboard/public/statics/locales/zh.json index a7b872537..d0900cdde 100644 --- a/dashboard/public/statics/locales/zh.json +++ b/dashboard/public/statics/locales/zh.json @@ -3027,7 +3027,9 @@ "expires": "过期时间", "serverNames": "证书服务器名称（有效 SNI）", "unknown": "未知", - "other": "其他" + "other": "其他", + "sniLabel": "SNI", + "sniDiscovered": "从 IP 发现" } }, "settings.cores.title": "核心", diff --git a/dashboard/src/features/core-editor/components/xray/reality-scan-dialog.tsx b/dashboard/src/features/core-editor/components/xray/reality-scan-dialog.tsx index e483360c3..3d2c09637 100644 --- a/dashboard/src/features/core-editor/components/xray/reality-scan-dialog.tsx +++ b/dashboard/src/features/core-editor/components/xray/reality-scan-dialog.tsx @@ -17,7 +17,6 @@ interface RealityScanDialogProps { open: boolean onOpenChange: (open: boolean) => void initialTarget?: string - initialSni?: string } const MAX_TARGETS = 25 @@ -148,6 +147,20 @@ function ScanResultDetail({ result }: { result: RealityScanResult }) { + {result.sni ? ( +
+ {t('coreEditor.realityScan.sniLabel', { defaultValue: 'SNI' })}: + + {result.sni} + + {result.sni_discovered ? ( + + {t('coreEditor.realityScan.sniDiscovered', { defaultValue: 'discovered from IP' })} + + ) : null} +
+ ) : null} + {result.reason ? ( @@ -347,10 +360,9 @@ function TargetsInput({ value, onChange, disabled, max }: { value: string[]; onC ) } -export function RealityScanDialog({ open, onOpenChange, initialTarget, initialSni }: RealityScanDialogProps) { +export function RealityScanDialog({ open, onOpenChange, initialTarget }: RealityScanDialogProps) { const { t } = useTranslation() const [targets, setTargets] = useState([]) - const [sni, setSni] = useState('') const [timeoutInput, setTimeoutInput] = useState('10') const [rows, setRows] = useState([]) const [expanded, setExpanded] = useState(null) @@ -368,12 +380,11 @@ export function RealityScanDialog({ open, onOpenChange, initialTarget, initialSn abortRef.current?.abort() abortRef.current = null setTargets(initialTarget?.trim() ? [initialTarget.trim()] : []) - setSni(initialSni?.trim() ?? '') setRows([]) setExpanded(null) setFeasibleOnly(false) setIsScanning(false) - }, [open, initialTarget, initialSni]) + }, [open, initialTarget]) useEffect(() => () => abortRef.current?.abort(), []) @@ -387,7 +398,6 @@ export function RealityScanDialog({ open, onOpenChange, initialTarget, initialSn abortRef.current = controller const parsedTimeout = Number(timeoutInput) const timeout = Number.isFinite(parsedTimeout) && parsedTimeout > 0 ? parsedTimeout : undefined - const useSni = list.length === 1 ? sni.trim() || undefined : undefined setExpanded(list.length === 1 ? list[0] : null) setFeasibleOnly(false) setRows(list.map(target => ({ target, status: 'pending' as RowStatus }))) @@ -403,7 +413,7 @@ export function RealityScanDialog({ open, onOpenChange, initialTarget, initialSn if (controller.signal.aborted) return patch(target, { status: 'scanning' }) try { - const res = await scanRealityTarget({ target, sni: useSni, timeout }, controller.signal) + const res = await scanRealityTarget({ target, timeout }, controller.signal) if (controller.signal.aborted) return patch(target, { status: 'done', result: res }) } catch (error) { @@ -467,15 +477,7 @@ export function RealityScanDialog({ open, onOpenChange, initialTarget, initialSn
- { - setTargets(next) - setSni('') - }} - disabled={isScanning} - max={MAX_TARGETS} - /> + {targets.length >= MAX_TARGETS ? (

{t('coreEditor.realityScan.maxTargets', { defaultValue: 'Up to {{max}} targets can be scanned.', max: MAX_TARGETS })}

) : null} diff --git a/dashboard/src/features/core-editor/components/xray/xray-inbounds-section.tsx b/dashboard/src/features/core-editor/components/xray/xray-inbounds-section.tsx index f09d17a3d..5e9c1a196 100644 --- a/dashboard/src/features/core-editor/components/xray/xray-inbounds-section.tsx +++ b/dashboard/src/features/core-editor/components/xray/xray-inbounds-section.tsx @@ -94,24 +94,6 @@ function securityFieldName(jsonKey: string): string { return `${SECURITY_FIELD_PREFIX}${jsonKey}` } -function firstConfiguredServerName(raw: unknown): string { - const s = typeof raw === 'string' ? raw : Array.isArray(raw) ? JSON.stringify(raw) : '' - const trimmed = s.trim() - if (!trimmed) return '' - let parts: string[] = [] - if (trimmed.startsWith('[')) { - try { - const parsed: unknown = JSON.parse(trimmed) - parts = Array.isArray(parsed) ? parsed.map(item => String(item ?? '').trim()) : [] - } catch { - parts = trimmed.split(/[\n,]/).map(part => part.trim()) - } - } else { - parts = trimmed.split(/[\n,]/).map(part => part.trim()) - } - return parts.find(p => p && !p.startsWith('*')) || '' -} - /** Plain English only — Xray REALITY / TLS / ECH field hints (not i18n). */ const INBOUND_SECURITY_PARITY_PLACEHOLDER: Readonly> = { dest: 'host:port for REALITY handshake target (e.g. www.microsoft.com:443)', @@ -922,7 +904,6 @@ export function XrayInboundsSection({ headerAddPulse, headerAddEpoch }: XrayInbo const [isGeneratingMldsa65, setIsGeneratingMldsa65] = useState(false) const [isRealityScanOpen, setIsRealityScanOpen] = useState(false) const [realityScanTarget, setRealityScanTarget] = useState('') - const [realityScanSni, setRealityScanSni] = useState('') const [echUsageOption, setEchUsageOption] = useState<'default' | 'required' | 'preferred'>('default') const [draftInbound, setDraftInbound] = useState(null) const [editOriginalInbound, setEditOriginalInbound] = useState(null) @@ -4049,7 +4030,6 @@ export function XrayInboundsSection({ headerAddPulse, headerAddEpoch }: XrayInbo onClick={() => { const destValue = form.getValues(securityFieldName('target')) setRealityScanTarget(typeof destValue === 'string' ? destValue : '') - setRealityScanSni(firstConfiguredServerName(form.getValues(securityFieldName('serverNames')))) setIsRealityScanOpen(true) }} className="h-10 w-full text-sm font-medium transition-all hover:shadow-md sm:h-11" @@ -5368,7 +5348,7 @@ export function XrayInboundsSection({ headerAddPulse, headerAddEpoch }: XrayInbo - + diff --git a/dashboard/src/service/reality-scan.ts b/dashboard/src/service/reality-scan.ts index 6e61a9aac..c692ce747 100644 --- a/dashboard/src/service/reality-scan.ts +++ b/dashboard/src/service/reality-scan.ts @@ -2,7 +2,6 @@ import { orvalFetcher } from './http' export interface RealityScanRequest { target: string - sni?: string | null timeout?: number | null } @@ -12,6 +11,7 @@ export interface RealityScanResult { ip: string | null port: number sni: string | null + sni_discovered: boolean feasible: boolean tls13: boolean tls_version: string | null diff --git a/tests/test_reality_scan_unit.py b/tests/test_reality_scan_unit.py index e6404f286..4e1c025b8 100644 --- a/tests/test_reality_scan_unit.py +++ b/tests/test_reality_scan_unit.py @@ -27,11 +27,6 @@ def test_parse_target_ok(target, expected): assert rs.parse_target(target) == expected -def test_parse_target_sni_override(): - host, port, sni = rs.parse_target("1.1.1.1:443", sni_override="example.org") - assert (host, port, sni) == ("1.1.1.1", 443, "example.org") - - @pytest.mark.parametrize("bad", ["", " ", "host:0", "host:70000", "host:abc"]) def test_parse_target_invalid(bad): with pytest.raises(RealityScanError): @@ -42,8 +37,6 @@ def test_parse_target_invalid(bad): def test_parse_target_rejects_control_chars(bad): with pytest.raises(RealityScanError): rs.parse_target(bad) - with pytest.raises(RealityScanError): - rs.parse_target("example.com", sni_override=bad) @pytest.mark.parametrize( @@ -205,6 +198,25 @@ def test_parse_certificate_handles_none(): assert rs._parse_certificate(None)["server_names"] == [] +def test_first_usable_name_prefers_common_name(): + der = _self_signed_der("cloudflare-dns.com", ["cloudflare-dns.com", "one.one.one.one"]) + assert rs._first_usable_name(der) == "cloudflare-dns.com" + + +def test_first_usable_name_skips_wildcard_cn_uses_san(): + der = _self_signed_der("*.example.com", ["*.example.com", "www.example.com"]) + assert rs._first_usable_name(der) == "www.example.com" + + +def test_first_usable_name_none_when_all_wildcard(): + der = _self_signed_der("*.example.com", ["*.example.com"]) + assert rs._first_usable_name(der) is None + + +def test_first_usable_name_none_when_no_cert(): + assert rs._first_usable_name(None) is None + + def _patch_probes(monkeypatch, *, tls, group, h3): monkeypatch.setattr(rs, "_tls_probe", lambda *a, **k: tls) monkeypatch.setattr(rs, "_group_probe", lambda *a, **k: group) @@ -223,6 +235,8 @@ def _patch_probes(monkeypatch, *, tls, group, h3): "server_names": ["example.com"], "latency_ms": 42, "reason": None, + "sni": "example.com", + "sni_discovered": False, } @@ -240,6 +254,15 @@ def test_scan_sync_feasible_when_group_unknown(monkeypatch): assert out["feasible"] is True +def test_scan_sync_carries_discovered_sni(monkeypatch): + tls = dict(_GOOD_TLS, sni="cloudflare-dns.com", sni_discovered=True) + _patch_probes(monkeypatch, tls=tls, group={"x25519": True, "post_quantum": True, "curve": "X25519MLKEM768"}, h3=False) + out = rs._scan_sync("1.0.0.1", "1.0.0.1", 443, None, 5) + assert out["sni"] == "cloudflare-dns.com" + assert out["sni_discovered"] is True + assert out["feasible"] is True + + def test_scan_sync_not_feasible_when_definitely_not_x25519(monkeypatch): _patch_probes(monkeypatch, tls=dict(_GOOD_TLS), group={"x25519": False, "post_quantum": False, "curve": "secp256r1"}, h3=False) out = rs._scan_sync("example.com", "93.184.216.34", 443, "example.com", 5) @@ -281,6 +304,17 @@ async def test_scan_reality_target_live_example(): assert result["h2"] is True assert result["cert_valid"] is True assert result["latency_ms"] is not None + assert result["sni"] == "example.com" + assert result["sni_discovered"] is False + + +@pytest.mark.skipif(os.environ.get("REALITY_SCAN_NETWORK_TEST") != "1", reason="network test opt-in") +@pytest.mark.asyncio +async def test_scan_reality_target_bare_ip_discovers_sni(): + result = await rs.scan_reality_target("1.0.0.1:443", timeout=8) + assert result["sni_discovered"] is True + assert result["sni"] + assert result["cert_valid"] is True def _frame(payload: bytes, rtype: int = 0x16) -> bytes: From d4c8f04897ea208d71238d5f592bdb701d358a05 Mon Sep 17 00:00:00 2001 From: parsa Date: Wed, 15 Jul 2026 14:03:27 -0700 Subject: [PATCH 2/4] fix(core): harden the REALITY scanner (handshake deadline, per-loop semaphore) - Bound the TLS handshake with an absolute deadline (non-blocking socket + select loop) so a slow-drip peer can no longer keep do_handshake() running forever, and run scans on a dedicated ThreadPoolExecutor so a lingering scan thread can never starve the event loop's shared pool (getaddrinfo / other to_thread work). - Key the concurrency semaphore per running event loop via a WeakKeyDictionary, removing the cross-loop "bound to a different event loop" RuntimeError. - Preserve the certificate reason when the permissive fallback also fails, handle UnicodeError from over-long DNS labels gracefully instead of a 502, and gate the discovered SNI through a hostname sanity check. --- app/utils/reality_scan.py | 154 ++++++++++++++++++++++-------- tests/test_reality_scan_unit.py | 163 +++++++++++++++++++++++++++++++- 2 files changed, 279 insertions(+), 38 deletions(-) diff --git a/app/utils/reality_scan.py b/app/utils/reality_scan.py index b7275e020..76b2624f0 100644 --- a/app/utils/reality_scan.py +++ b/app/utils/reality_scan.py @@ -1,10 +1,13 @@ import asyncio import ipaddress import os +import select import socket import ssl import struct import time +import weakref +from concurrent.futures import ThreadPoolExecutor from cryptography import x509 from cryptography.hazmat.primitives.asymmetric import x25519 as _x25519 @@ -21,14 +24,26 @@ DNS_TIMEOUT = 5.0 MAX_CONCURRENT_SCANS = 4 -_scan_semaphore: "asyncio.Semaphore | None" = None +_scan_semaphores: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Semaphore]" = weakref.WeakKeyDictionary() + +_scan_executor: "ThreadPoolExecutor | None" = None def _get_scan_semaphore() -> "asyncio.Semaphore": - global _scan_semaphore - if _scan_semaphore is None: - _scan_semaphore = asyncio.Semaphore(MAX_CONCURRENT_SCANS) - return _scan_semaphore + loop = asyncio.get_running_loop() + sem = _scan_semaphores.get(loop) + if sem is None: + sem = asyncio.Semaphore(MAX_CONCURRENT_SCANS) + _scan_semaphores[loop] = sem + return sem + + +def _get_scan_executor() -> "ThreadPoolExecutor": + global _scan_executor + if _scan_executor is None: + _scan_executor = ThreadPoolExecutor(max_workers=MAX_CONCURRENT_SCANS + 2, thread_name_prefix="reality-scan") + return _scan_executor + GROUP_X25519 = 0x001D GROUP_X25519MLKEM768 = 0x11EC @@ -235,6 +250,17 @@ def _parse_certificate(der: bytes | None) -> dict: return out +def _looks_like_hostname(name: str) -> bool: + if not name or len(name) > 253 or _has_control_chars(name): + return False + if any(ch.isspace() for ch in name): + return False + labels = name.rstrip(".").split(".") + if len(labels) < 2: + return False + return all(0 < len(label) <= 63 for label in labels) + + def _first_usable_name(der: bytes | None) -> str | None: if not der: return None @@ -243,19 +269,63 @@ def _first_usable_name(der: bytes | None) -> str | None: except Exception: return None cn = _name_common_name(cert.subject) - if cn and not cn.startswith("*."): - return cn.strip() + if cn: + cn = cn.strip() + if not cn.startswith("*.") and _looks_like_hostname(cn): + return cn try: san = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME) for name in san.value.get_values_for_type(x509.DNSName): name = name.strip() - if name and not name.startswith("*."): + if name and not name.startswith("*.") and _looks_like_hostname(name): return name except Exception: pass return None +def _wait_io(tls: ssl.SSLSocket, deadline: float, *, want_read: bool) -> bool: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + if want_read: + ready, _, _ = select.select([tls], [], [], remaining) + else: + _, ready, _ = select.select([], [tls], [], remaining) + return bool(ready) + + +def _drive_handshake(tls: ssl.SSLSocket, deadline: float) -> None: + while True: + try: + tls.do_handshake() + return + except ssl.SSLWantReadError: + if not _wait_io(tls, deadline, want_read=True): + raise TimeoutError("TLS handshake timed out") + except ssl.SSLWantWriteError: + if not _wait_io(tls, deadline, want_read=False): + raise TimeoutError("TLS handshake timed out") + + +def _tls_wrap_with_deadline(ctx: ssl.SSLContext, ip: str, port: int, server_hostname: str | None, timeout: float) -> tuple[ssl.SSLSocket, float]: + started = time.monotonic() + deadline = started + timeout + sock = socket.create_connection((ip, port), timeout=timeout) + try: + sock.setblocking(False) + tls = ctx.wrap_socket(sock, server_hostname=server_hostname, do_handshake_on_connect=False) + except BaseException: + sock.close() + raise + try: + _drive_handshake(tls, deadline) + except BaseException: + tls.close() + raise + return tls, (time.monotonic() - started) * 1000.0 + + def _make_verify_ctx() -> ssl.SSLContext: ctx = ssl.create_default_context() ctx.minimum_version = ssl.TLSVersion.TLSv1_2 @@ -294,11 +364,9 @@ def _tls_probe(ip: str, port: int, sni: str | None, timeout: float) -> dict: der: bytes | None = None def _handshake(ctx: ssl.SSLContext, server_hostname: str | None) -> tuple[str | None, str | None, bytes | None, float]: - started = time.monotonic() - with socket.create_connection((ip, port), timeout=timeout) as sock: - with ctx.wrap_socket(sock, server_hostname=server_hostname) as tls: - latency = (time.monotonic() - started) * 1000.0 - return tls.version(), tls.selected_alpn_protocol(), tls.getpeercert(binary_form=True), latency + tls, latency = _tls_wrap_with_deadline(ctx, ip, port, server_hostname, timeout) + with tls: + return tls.version(), tls.selected_alpn_protocol(), tls.getpeercert(binary_form=True), latency try: if sni is not None: @@ -308,8 +376,12 @@ def _handshake(ctx: ssl.SSLContext, server_hostname: str | None) -> tuple[str | result["latency_ms"] = round(latency) except ssl.SSLCertVerificationError as exc: result["reason"] = f"Certificate did not validate: {getattr(exc, 'verify_message', None) or exc}" - version, alpn, der, latency = _handshake(_make_permissive_ctx(), sni) - result["latency_ms"] = round(latency) + try: + version, alpn, der, latency = _handshake(_make_permissive_ctx(), sni) + result["latency_ms"] = round(latency) + except (ssl.SSLError, socket.timeout, TimeoutError, OSError, UnicodeError) as exc2: + logger.debug("reality-scan: permissive fallback handshake failed for %s: %s", sni, exc2) + return result else: version, alpn, der, latency = _handshake(_make_permissive_ctx(), None) result["latency_ms"] = round(latency) @@ -325,7 +397,7 @@ def _handshake(ctx: ssl.SSLContext, server_hostname: str | None) -> tuple[str | result["latency_ms"] = round(latency) except ssl.SSLCertVerificationError as exc: result["reason"] = f"Certificate did not validate: {getattr(exc, 'verify_message', None) or exc}" - except (ssl.SSLError, socket.timeout, TimeoutError, OSError) as exc: + except (ssl.SSLError, socket.timeout, TimeoutError, OSError, UnicodeError) as exc: result["reason"] = f"Certificate re-validation failed: {exc}" except (socket.timeout, TimeoutError): result["reason"] = "Connection timed out." @@ -336,6 +408,9 @@ def _handshake(ctx: ssl.SSLContext, server_hostname: str | None) -> tuple[str | except (ConnectionRefusedError, ConnectionResetError, OSError) as exc: result["reason"] = f"Connection failed: {exc}" return result + except UnicodeError as exc: + result["reason"] = f"Server name could not be encoded for TLS: {exc}" + return result if version is None: return result @@ -526,27 +601,27 @@ def _h3_probe(host: str, ip: str, port: int, sni: str | None, timeout: float) -> ctx.verify_mode = ssl.CERT_NONE ctx.set_alpn_protocols(["http/1.1"]) request_host = sni or host + tls, _ = _tls_wrap_with_deadline(ctx, ip, port, sni, timeout) deadline = time.monotonic() + timeout - with socket.create_connection((ip, port), timeout=timeout) as sock: - with ctx.wrap_socket(sock, server_hostname=sni) as tls: - tls.settimeout(timeout) - request = ( - f"GET / HTTP/1.1\r\nHost: {request_host}\r\n" - "User-Agent: PasarGuard-RealityScan/1.0\r\nAccept: */*\r\nConnection: close\r\n\r\n" - ) - tls.sendall(request.encode("ascii", "ignore")) - data = b"" - while len(data) < 32768: - remaining = deadline - time.monotonic() - if remaining <= 0: - break - tls.settimeout(remaining) - chunk = tls.recv(4096) - if not chunk: - break - data += chunk - if b"\r\n\r\n" in data: - break + with tls: + tls.settimeout(timeout) + request = ( + f"GET / HTTP/1.1\r\nHost: {request_host}\r\n" + "User-Agent: PasarGuard-RealityScan/1.0\r\nAccept: */*\r\nConnection: close\r\n\r\n" + ) + tls.sendall(request.encode("ascii", "ignore")) + data = b"" + while len(data) < 32768: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + tls.settimeout(remaining) + chunk = tls.recv(4096) + if not chunk: + break + data += chunk + if b"\r\n\r\n" in data: + break header_blob = data.split(b"\r\n\r\n", 1)[0].decode("latin-1", "ignore") for line in header_blob.split("\r\n"): lower = line.lower() @@ -606,7 +681,12 @@ async def scan_reality_target(target: str, timeout: float | None = None) -> dict clamped = _clamp_timeout(timeout) async with _get_scan_semaphore(): ip = await _resolve_public_ip_async(host, min(clamped, DNS_TIMEOUT)) + loop = asyncio.get_running_loop() + executor = _get_scan_executor() try: - return await asyncio.wait_for(asyncio.to_thread(_scan_sync, host, ip, port, sni, clamped), timeout=clamped * 6 + 15) + return await asyncio.wait_for( + loop.run_in_executor(executor, _scan_sync, host, ip, port, sni, clamped), + timeout=clamped * 6 + 15, + ) except TimeoutError: raise RealityScanError("Scan timed out.") diff --git a/tests/test_reality_scan_unit.py b/tests/test_reality_scan_unit.py index 4e1c025b8..306b5ace4 100644 --- a/tests/test_reality_scan_unit.py +++ b/tests/test_reality_scan_unit.py @@ -425,7 +425,8 @@ async def test_scan_concurrency_is_capped(monkeypatch): import threading import time as _time - rs._scan_semaphore = None + rs._scan_semaphores.clear() + rs._scan_executor = None lock = threading.Lock() state = {"live": 0, "peak": 0} resolver = {"live": 0, "peak": 0} @@ -452,3 +453,163 @@ def fake_sync(*a, **k): await asyncio.gather(*[rs.scan_reality_target("example.com:443") for _ in range(12)]) assert state["peak"] <= rs.MAX_CONCURRENT_SCANS assert resolver["peak"] <= rs.MAX_CONCURRENT_SCANS + + + + +class _FakeTLS: + + def __init__(self, version="TLSv1.3", alpn="h2", der=b"DER"): + self._version, self._alpn, self._der = version, alpn, der + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def version(self): + return self._version + + def selected_alpn_protocol(self): + return self._alpn + + def getpeercert(self, binary_form=False): + return self._der + + +def test_looks_like_hostname_cases(): + good = ["example.com", "www.a.b.co", "one.one.one.one", "cloudflare-dns.com", "example.com."] + bad = ["", "localhost", "a b.com", "a" * 70 + ".com", "x." + "a" * 64, "h\x01st.com"] + assert all(rs._looks_like_hostname(n) for n in good), [n for n in good if not rs._looks_like_hostname(n)] + assert not any(rs._looks_like_hostname(n) for n in bad), [n for n in bad if rs._looks_like_hostname(n)] + + +def test_first_usable_name_skips_non_hostname_cn_uses_san(): + der = _self_signed_der("localhost", ["localhost", "good.example.com"]) + assert rs._first_usable_name(der) == "good.example.com" + + +def test_first_usable_name_none_for_org_string_cn_no_san(): + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Some Org CA")]) + now = datetime.datetime.now(datetime.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=90)) + .sign(key, hashes.SHA256()) + ) + assert rs._first_usable_name(cert.public_bytes(serialization.Encoding.DER)) is None + + +def test_wait_io_false_on_select_timeout(monkeypatch): + import time as _t + + monkeypatch.setattr(rs.select, "select", lambda r, w, x, t: ([], [], [])) + assert rs._wait_io(object(), deadline=_t.monotonic() + 5, want_read=True) is False + + +def test_drive_handshake_times_out_when_deadline_passed(): + import time as _t + + class T: + def do_handshake(self): + raise rs.ssl.SSLWantReadError() + + with pytest.raises(TimeoutError): + rs._drive_handshake(T(), deadline=_t.monotonic() - 1) + + +def test_drive_handshake_retries_then_succeeds(monkeypatch): + import time as _t + + calls = {"n": 0} + + class T: + def do_handshake(self): + calls["n"] += 1 + if calls["n"] == 1: + raise rs.ssl.SSLWantReadError() + + monkeypatch.setattr(rs.select, "select", lambda r, w, x, t: ([object()], [], [])) + rs._drive_handshake(T(), deadline=_t.monotonic() + 5) + assert calls["n"] == 2 + + +def test_get_scan_executor_is_bounded(): + rs._scan_executor = None + ex = rs._get_scan_executor() + assert ex._max_workers == rs.MAX_CONCURRENT_SCANS + 2 + rs._scan_executor = None + + +@pytest.mark.asyncio +async def test_get_scan_semaphore_same_within_loop(): + rs._scan_semaphores.clear() + assert rs._get_scan_semaphore() is rs._get_scan_semaphore() + + +def test_get_scan_semaphore_distinct_across_sequential_loops(): + import asyncio + + async def get(): + return rs._get_scan_semaphore() + + rs._scan_semaphores.clear() + s1 = asyncio.run(get()) + s2 = asyncio.run(get()) + assert s1 is not s2 + + +def test_get_scan_semaphore_requires_running_loop(): + with pytest.raises(RuntimeError): + rs._get_scan_semaphore() + + +def test_tls_probe_hostname_fallback_preserves_cert_reason(monkeypatch): + calls = {"n": 0} + + def fake_wrap(ctx, ip, port, server_hostname, timeout): + calls["n"] += 1 + if calls["n"] == 1: + raise rs.ssl.SSLCertVerificationError("hostname mismatch") + raise OSError("connection reset") + + monkeypatch.setattr(rs, "_tls_wrap_with_deadline", fake_wrap) + out = rs._tls_probe("1.2.3.4", 443, "example.com", 2) + assert out["cert_valid"] is False + assert out["reason"].startswith("Certificate did not validate") + + +def test_tls_probe_bare_ip_revalidation_unicodeerror_is_graceful(monkeypatch): + calls = {"n": 0} + + def fake_wrap(ctx, ip, port, server_hostname, timeout): + calls["n"] += 1 + if calls["n"] == 1: + return _FakeTLS(), 12.0 + raise UnicodeError("label empty or too long") + + monkeypatch.setattr(rs, "_tls_wrap_with_deadline", fake_wrap) + monkeypatch.setattr(rs, "_first_usable_name", lambda der: "cloudflare-dns.com") + out = rs._tls_probe("1.0.0.1", 443, None, 2) + assert out["sni"] == "cloudflare-dns.com" + assert out["sni_discovered"] is True + assert out["cert_valid"] is False + assert out["tls_version"] == "1.3" + assert out["reason"].startswith("Certificate re-validation failed") + + +def test_tls_probe_outer_unicodeerror_on_target_sni(monkeypatch): + def fake_wrap(ctx, ip, port, server_hostname, timeout): + raise UnicodeError("label too long") + + monkeypatch.setattr(rs, "_tls_wrap_with_deadline", fake_wrap) + out = rs._tls_probe("1.2.3.4", 443, "a" * 70 + ".com", 2) + assert out["tls_version"] is None + assert out["reason"].startswith("Server name could not be encoded for TLS") From 8ee5e489ff8b7ed8de60bc75fcffcc22b73ad010 Mon Sep 17 00:00:00 2001 From: parsa Date: Wed, 15 Jul 2026 15:12:41 -0700 Subject: [PATCH 3/4] fix(core): address PR #700 review on the REALITY scanner Prefer non-wildcard DNS SANs over the certificate CN when discovering a bare-IP target's server name, and fall back to the CN only when the certificate carries no DNS SANs. This matches OpenSSL hostname verification, which ignores the CN whenever any dNSName SAN is present, so a CN-derived name no longer fails re-validation on an otherwise valid target. Require a confirmed X25519-family key exchange for feasibility instead of only rejecting a confirmed non-X25519 curve. A group probe that cannot determine the curve now marks the target not feasible with an explicit reason rather than passing on an unverified assumption. --- app/utils/reality_scan.py | 31 ++++++++++++++++++------------ tests/test_reality_scan_unit.py | 34 ++++++++++++++++++++++++++++----- 2 files changed, 48 insertions(+), 17 deletions(-) diff --git a/app/utils/reality_scan.py b/app/utils/reality_scan.py index 76b2624f0..44d52c51c 100644 --- a/app/utils/reality_scan.py +++ b/app/utils/reality_scan.py @@ -268,19 +268,21 @@ def _first_usable_name(der: bytes | None) -> str | None: cert = x509.load_der_x509_certificate(der) except Exception: return None - cn = _name_common_name(cert.subject) - if cn: - cn = cn.strip() - if not cn.startswith("*.") and _looks_like_hostname(cn): - return cn try: san = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME) - for name in san.value.get_values_for_type(x509.DNSName): - name = name.strip() - if name and not name.startswith("*.") and _looks_like_hostname(name): - return name + dns_names = list(san.value.get_values_for_type(x509.DNSName)) except Exception: - pass + dns_names = [] + for name in dns_names: + name = name.strip() + if name and not name.startswith("*.") and _looks_like_hostname(name): + return name + if not dns_names: + cn = _name_common_name(cert.subject) + if cn: + cn = cn.strip() + if not cn.startswith("*.") and _looks_like_hostname(cn): + return cn return None @@ -671,8 +673,13 @@ def _scan_sync(host: str, ip: str, port: int, sni: str | None, timeout: float) - result["curve"] = group["curve"] result["h3"] = _h3_probe(host, ip, port, effective_sni, timeout) - definitely_not_x25519 = group["x25519"] is False and group["post_quantum"] is False and group["curve"] is not None - result["feasible"] = bool(result["tls13"] and result["h2"] and result["cert_valid"] and not definitely_not_x25519) + base_ok = bool(result["tls13"] and result["h2"] and result["cert_valid"]) + result["feasible"] = base_ok and result["x25519"] is True + if base_ok and result["x25519"] is not True and not result["reason"]: + if result["x25519"] is False: + result["reason"] = f"Key exchange is {result['curve'] or 'not X25519'}; REALITY needs X25519 or X25519MLKEM768." + else: + result["reason"] = "Could not confirm an X25519 key exchange." return result diff --git a/tests/test_reality_scan_unit.py b/tests/test_reality_scan_unit.py index 306b5ace4..8f7203b31 100644 --- a/tests/test_reality_scan_unit.py +++ b/tests/test_reality_scan_unit.py @@ -198,9 +198,31 @@ def test_parse_certificate_handles_none(): assert rs._parse_certificate(None)["server_names"] == [] -def test_first_usable_name_prefers_common_name(): - der = _self_signed_der("cloudflare-dns.com", ["cloudflare-dns.com", "one.one.one.one"]) - assert rs._first_usable_name(der) == "cloudflare-dns.com" +def test_first_usable_name_prefers_san_over_cn(): + der = _self_signed_der("legacy-cn.example", ["real.example.com", "one.one.one.one"]) + assert rs._first_usable_name(der) == "real.example.com" + + +def test_first_usable_name_ignores_cn_when_san_present_but_unusable(): + der = _self_signed_der("apex.example.com", ["*.example.com"]) + assert rs._first_usable_name(der) is None + + +def test_first_usable_name_falls_back_to_cn_without_san(): + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "cn-only.example.com")]) + now = datetime.datetime.now(datetime.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=90)) + .sign(key, hashes.SHA256()) + ) + assert rs._first_usable_name(cert.public_bytes(serialization.Encoding.DER)) == "cn-only.example.com" def test_first_usable_name_skips_wildcard_cn_uses_san(): @@ -248,10 +270,11 @@ def test_scan_sync_feasible_when_all_pass(monkeypatch): assert out["h3"] is True -def test_scan_sync_feasible_when_group_unknown(monkeypatch): +def test_scan_sync_not_feasible_when_group_unknown(monkeypatch): _patch_probes(monkeypatch, tls=dict(_GOOD_TLS), group={"x25519": None, "post_quantum": None, "curve": None}, h3=False) out = rs._scan_sync("example.com", "93.184.216.34", 443, "example.com", 5) - assert out["feasible"] is True + assert out["feasible"] is False + assert "X25519" in out["reason"] def test_scan_sync_carries_discovered_sni(monkeypatch): @@ -267,6 +290,7 @@ def test_scan_sync_not_feasible_when_definitely_not_x25519(monkeypatch): _patch_probes(monkeypatch, tls=dict(_GOOD_TLS), group={"x25519": False, "post_quantum": False, "curve": "secp256r1"}, h3=False) out = rs._scan_sync("example.com", "93.184.216.34", 443, "example.com", 5) assert out["feasible"] is False + assert "secp256r1" in out["reason"] def test_scan_sync_not_feasible_without_tls13(monkeypatch): From c533339be2e968f56ef1525523ff96db4b16c33e Mon Sep 17 00:00:00 2001 From: parsa Date: Wed, 15 Jul 2026 15:24:47 -0700 Subject: [PATCH 4/4] test(core): reuse the cert helper for no-SAN scanner cases Let _self_signed_der skip the SubjectAlternativeName extension when no SANs are given, and collapse the three inline no-SAN certificate builders onto it, addressing the PR #700 review nitpick. --- tests/test_reality_scan_unit.py | 53 +++++---------------------------- 1 file changed, 7 insertions(+), 46 deletions(-) diff --git a/tests/test_reality_scan_unit.py b/tests/test_reality_scan_unit.py index 8f7203b31..70df82d96 100644 --- a/tests/test_reality_scan_unit.py +++ b/tests/test_reality_scan_unit.py @@ -171,7 +171,7 @@ def _self_signed_der(cn: str, sans: list[str]) -> bytes: key = rsa.generate_private_key(public_exponent=65537, key_size=2048) name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Test Org")]) now = datetime.datetime.now(datetime.timezone.utc) - cert = ( + builder = ( x509.CertificateBuilder() .subject_name(name) .issuer_name(name) @@ -179,10 +179,10 @@ def _self_signed_der(cn: str, sans: list[str]) -> bytes: .serial_number(x509.random_serial_number()) .not_valid_before(now - datetime.timedelta(days=1)) .not_valid_after(now + datetime.timedelta(days=90)) - .add_extension(x509.SubjectAlternativeName([x509.DNSName(s) for s in sans]), critical=False) - .sign(key, hashes.SHA256()) ) - return cert.public_bytes(serialization.Encoding.DER) + if sans: + builder = builder.add_extension(x509.SubjectAlternativeName([x509.DNSName(s) for s in sans]), critical=False) + return builder.sign(key, hashes.SHA256()).public_bytes(serialization.Encoding.DER) def test_parse_certificate_extracts_sans_and_filters_wildcards(): @@ -209,20 +209,7 @@ def test_first_usable_name_ignores_cn_when_san_present_but_unusable(): def test_first_usable_name_falls_back_to_cn_without_san(): - key = rsa.generate_private_key(public_exponent=65537, key_size=2048) - name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "cn-only.example.com")]) - now = datetime.datetime.now(datetime.timezone.utc) - cert = ( - x509.CertificateBuilder() - .subject_name(name) - .issuer_name(name) - .public_key(key.public_key()) - .serial_number(x509.random_serial_number()) - .not_valid_before(now - datetime.timedelta(days=1)) - .not_valid_after(now + datetime.timedelta(days=90)) - .sign(key, hashes.SHA256()) - ) - assert rs._first_usable_name(cert.public_bytes(serialization.Encoding.DER)) == "cn-only.example.com" + assert rs._first_usable_name(_self_signed_der("cn-only.example.com", [])) == "cn-only.example.com" def test_first_usable_name_skips_wildcard_cn_uses_san(): @@ -418,20 +405,7 @@ def test_group_probe_non_x25519_group(monkeypatch): def test_parse_certificate_without_sans(): - key = rsa.generate_private_key(public_exponent=65537, key_size=2048) - name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "no-san.example")]) - now = datetime.datetime.now(datetime.timezone.utc) - cert = ( - x509.CertificateBuilder() - .subject_name(name) - .issuer_name(name) - .public_key(key.public_key()) - .serial_number(x509.random_serial_number()) - .not_valid_before(now - datetime.timedelta(days=1)) - .not_valid_after(now + datetime.timedelta(days=90)) - .sign(key, hashes.SHA256()) - ) - out = rs._parse_certificate(cert.public_bytes(serialization.Encoding.DER)) + out = rs._parse_certificate(_self_signed_der("no-san.example", [])) assert out["cert_subject"] == "no-san.example" assert out["server_names"] == [] @@ -515,20 +489,7 @@ def test_first_usable_name_skips_non_hostname_cn_uses_san(): def test_first_usable_name_none_for_org_string_cn_no_san(): - key = rsa.generate_private_key(public_exponent=65537, key_size=2048) - name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Some Org CA")]) - now = datetime.datetime.now(datetime.timezone.utc) - cert = ( - x509.CertificateBuilder() - .subject_name(name) - .issuer_name(name) - .public_key(key.public_key()) - .serial_number(x509.random_serial_number()) - .not_valid_before(now - datetime.timedelta(days=1)) - .not_valid_after(now + datetime.timedelta(days=90)) - .sign(key, hashes.SHA256()) - ) - assert rs._first_usable_name(cert.public_bytes(serialization.Encoding.DER)) is None + assert rs._first_usable_name(_self_signed_der("Some Org CA", [])) is None def test_wait_io_false_on_select_timeout(monkeypatch):