mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-11-05 20:52:22 +01:00
fix(companion): honor configured pairing address (#6060)
* fix(companion): honor configured pairing origin * fix(companion): keep configured pairing on v1 LAN contract * fix(companion): reject numeric pairing hosts --------- Co-authored-by: RaresKeY <158580472+RaresKeY@users.noreply.github.com>
This commit is contained in:
co-authored by
RaresKeY
parent
18991d6f67
commit
9c71948376
@@ -6,11 +6,14 @@ units so the route layer stays thin and the logic is directly testable.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
import socket
|
||||
import uuid
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
import bcrypt
|
||||
|
||||
@@ -20,6 +23,102 @@ PAIRING_VERSION = 1
|
||||
COMPANION_SCOPE = "chat"
|
||||
|
||||
|
||||
_COMPANION_IPV4_NETWORKS = tuple(
|
||||
ipaddress.ip_network(cidr)
|
||||
for cidr in (
|
||||
"10.0.0.0/8",
|
||||
"100.64.0.0/10",
|
||||
"127.0.0.0/8",
|
||||
"169.254.0.0/16",
|
||||
"172.16.0.0/12",
|
||||
"192.168.0.0/16",
|
||||
)
|
||||
)
|
||||
_DNS_LABEL_RE = re.compile(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\Z")
|
||||
|
||||
|
||||
def _valid_companion_client_host(host: str) -> bool:
|
||||
"""Match the host forms supported by the current v1 Expo client."""
|
||||
if not host or len(host) > 253 or not host.isascii() or "%" in host:
|
||||
return False
|
||||
|
||||
try:
|
||||
address = ipaddress.ip_address(host)
|
||||
except ValueError:
|
||||
labels = host.split(".")
|
||||
if any(not _DNS_LABEL_RE.fullmatch(label) for label in labels):
|
||||
return False
|
||||
if any(label.startswith("xn--") for label in labels):
|
||||
return False
|
||||
# WHATWG URL parsers treat a decimal or ``0x`` single-label hostname
|
||||
# as an IPv4 number even though Python's strict ``ipaddress`` parser
|
||||
# rejects that spelling. The v1 client interpolates this host back
|
||||
# into a URL, so accepting e.g. ``134744072`` would make the phone send
|
||||
# its bearer token to public 8.8.8.8. Keep DNS labels unambiguous.
|
||||
if len(labels) == 1 and (
|
||||
labels[0].isdigit()
|
||||
or re.fullmatch(r"0x[0-9a-f]*", labels[0]) is not None
|
||||
):
|
||||
return False
|
||||
return len(labels) == 1 or (len(labels) >= 2 and labels[-1] == "local")
|
||||
|
||||
return isinstance(address, ipaddress.IPv4Address) and any(
|
||||
address in network for network in _COMPANION_IPV4_NETWORKS
|
||||
)
|
||||
|
||||
|
||||
def parse_companion_base_url(value: str) -> tuple[str, int]:
|
||||
"""Validate a v1 companion address and return its legacy (host, port).
|
||||
|
||||
The deployed client understands only HTTP plus a LAN-style host and port.
|
||||
Reject anything outside that exact contract instead of advertising a URL
|
||||
the client would reject, downgrade, or interpret differently.
|
||||
"""
|
||||
if not isinstance(value, str) or not value:
|
||||
raise ValueError("COMPANION_BASE_URL must be a canonical HTTP LAN origin")
|
||||
if not value.isascii():
|
||||
raise ValueError("COMPANION_BASE_URL must contain only ASCII characters")
|
||||
if any(
|
||||
ord(char) <= 32 or ord(char) == 127 or char in {"\\", "%"}
|
||||
for char in value
|
||||
):
|
||||
raise ValueError(
|
||||
"COMPANION_BASE_URL contains a forbidden character"
|
||||
)
|
||||
|
||||
try:
|
||||
parsed = urlsplit(value)
|
||||
port = parsed.port
|
||||
except ValueError as exc:
|
||||
raise ValueError("COMPANION_BASE_URL must be a valid HTTP LAN origin") from exc
|
||||
|
||||
host = parsed.hostname
|
||||
if parsed.scheme.lower() != "http" or not parsed.netloc or not host:
|
||||
raise ValueError("COMPANION_BASE_URL must be a canonical HTTP LAN origin")
|
||||
if parsed.username is not None or parsed.password is not None:
|
||||
raise ValueError("COMPANION_BASE_URL must not contain credentials")
|
||||
if parsed.path or parsed.query or parsed.fragment:
|
||||
raise ValueError("COMPANION_BASE_URL must not contain a path, query, or fragment")
|
||||
if port is not None and not 1 <= port <= 65535:
|
||||
raise ValueError("COMPANION_BASE_URL port must be between 1 and 65535")
|
||||
if not _valid_companion_client_host(host):
|
||||
raise ValueError("COMPANION_BASE_URL host is not supported by companion v1")
|
||||
|
||||
netloc = f"{host}:{port}" if port is not None else host
|
||||
origin = f"http://{netloc}"
|
||||
if value != origin:
|
||||
raise ValueError("COMPANION_BASE_URL must be a canonical HTTP LAN origin")
|
||||
return host, port or 80
|
||||
|
||||
|
||||
def configured_companion_origin() -> tuple[str, int] | None:
|
||||
"""Return the validated operator-configured v1 address, if any."""
|
||||
value = os.environ.get("COMPANION_BASE_URL")
|
||||
if value is None or value == "":
|
||||
return None
|
||||
return parse_companion_base_url(value)
|
||||
|
||||
|
||||
def default_port() -> int:
|
||||
"""Best guess at the port the server is reachable on. Callers that know the
|
||||
real request port should pass it explicitly."""
|
||||
|
||||
+13
-4
@@ -200,19 +200,27 @@ def setup_companion_routes() -> APIRouter:
|
||||
the code works immediately, no restart. `?format=json` returns the
|
||||
payload for an in-app pairing screen."""
|
||||
require_admin(request)
|
||||
try:
|
||||
configured_origin = _pairing.configured_companion_origin()
|
||||
except ValueError as exc:
|
||||
raise HTTPException(500, str(exc)) from None
|
||||
owner = get_current_user(request)
|
||||
invalidate = getattr(request.app.state, "invalidate_token_cache", None)
|
||||
token_id, raw_token = mint_pairing_token(owner, invalidate)
|
||||
|
||||
hosts = _pairing.lan_ip_candidates()
|
||||
host = hosts[0] if hosts else "127.0.0.1"
|
||||
port = request.url.port or _pairing.default_port()
|
||||
if configured_origin:
|
||||
host, port = configured_origin
|
||||
hosts = [host]
|
||||
else:
|
||||
hosts = _pairing.lan_ip_candidates()
|
||||
host = hosts[0] if hosts else "127.0.0.1"
|
||||
port = request.url.port or _pairing.default_port()
|
||||
payload = _pairing.pairing_payload(host, port, raw_token)
|
||||
qr = _pairing.pairing_qr_png_data_uri(payload)
|
||||
qr_ok = bool(qr and qr.startswith("data:image/png;base64,"))
|
||||
|
||||
if (request.query_params.get("format") or "").lower() == "json":
|
||||
return {
|
||||
response = {
|
||||
"host": host,
|
||||
"port": port,
|
||||
"token": raw_token,
|
||||
@@ -221,6 +229,7 @@ def setup_companion_routes() -> APIRouter:
|
||||
"payload": payload,
|
||||
"qr": qr if qr_ok else None,
|
||||
}
|
||||
return response
|
||||
|
||||
import json as _json
|
||||
payload_json = _json.dumps(payload, separators=(",", ":"))
|
||||
|
||||
Reference in New Issue
Block a user