diff --git a/adju-phone-camera/.gitignore b/adju-phone-camera/.gitignore new file mode 100644 index 00000000..7f4a2e74 --- /dev/null +++ b/adju-phone-camera/.gitignore @@ -0,0 +1,6 @@ +.venv/ +__pycache__/ +.pytest_cache/ +build/ +dist/ +*.pem diff --git a/adju-phone-camera/README.md b/adju-phone-camera/README.md new file mode 100644 index 00000000..d0686a72 --- /dev/null +++ b/adju-phone-camera/README.md @@ -0,0 +1,80 @@ +# ADJU PHONE CAMERA + +A Windows desktop app that receives an Android or iPhone camera feed over your local Wi-Fi. The app uses a private six-digit pairing PIN, a local WebSocket for signaling, and a direct WebRTC connection for video. It does not use cloud services, STUN/TURN servers, or an external video relay. + +## Requirements + +- Windows 10 or 11, 64-bit, and Python 3.10 or later. +- PC and phone on the same Wi-Fi/LAN; guest-network isolation can prevent them from seeing each other. +- A modern Android or iPhone browser. Camera permission is requested only after tapping **Start camera** on the phone. +- For camera access from a phone, use trusted HTTPS as described below. A plain `http://192.168.x.x` LAN URL is not a secure context and is not sufficient for `getUserMedia` on current Android browsers or Safari. + +## Install and run + +Open PowerShell in this directory: + +```powershell +py -3 -m venv .venv +.\.venv\Scripts\Activate.ps1 +python -m pip install --upgrade pip +python -m pip install -r requirements.txt +python main.py +``` + +The app automatically creates a private certificate authority and a short-lived HTTPS certificate for the PC's current LAN IP. On first run, it shows: + +- An HTTPS QR code for pairing and camera use on port `8765`. +- A certificate download link over the local network on port `8766`. +- The CA certificate's SHA-256 fingerprint, which must match the phone certificate before you trust it. + +First, install and trust the CA certificate on the phone as described below. Then scan the HTTPS QR code, enter the six-digit PIN from the PC, choose a camera and resolution, and tap **Start camera**. Approve the browser's camera prompt. On the desktop, the preview shows incoming video and FPS; use **Save screenshot** or **Fullscreen preview** as needed. The HTTP listener on port `8766` only serves the public CA certificate; pairing and signaling are available over HTTPS/WSS on port `8765`. + +The available video targets are 480p, 720p, and 1080p. Actual resolution and frame rate depend on the phone, browser, Wi-Fi, and operating system. Switch between front and back cameras on the phone page; stop the stream first to change settings. + +## Install and trust the local certificate on the phone + +The app stores its private CA key only on this PC under `%LOCALAPPDATA%\ADJU PHONE CAMERA\certificates`. It creates a server certificate for the current private LAN IP. Only the public CA certificate is downloadable; the private CA key is never served or bundled. A new server certificate is issued if the LAN IP changes, using the same CA. + +1. Start the desktop server. On the phone, open the **CA download** link shown by the app, for example `http://192.168.1.20:8766/ca.cer`. This separate HTTP endpoint serves only the public certificate, not the camera page or signaling. +2. Before installing it, compare the certificate's SHA-256 fingerprint shown in the desktop app with the certificate details on the phone wherever the OS exposes its fingerprint. Only install it if the identity matches and you trust this PC's CA. +3. Install the downloaded certificate specifically as a **CA/root certificate**, then explicitly enable trust for it: + - **Android:** Settings > Security & privacy (names vary) > More security settings > Encryption & credentials > Install a certificate > CA certificate. Confirm the OS trust warning only after checking the fingerprint. + - **iPhone/iPad:** Open the downloaded configuration profile in Settings and install it. Then open Settings > General > About > Certificate Trust Settings and enable full trust for this root certificate. Menu names vary by iOS version. +4. Return to the app and scan its **HTTPS** QR code, such as `https://192.168.1.20:8765/`. The browser must show a secure page without a certificate warning. Enter the pairing PIN, tap **Start camera**, and grant camera permission. + +If the phone OS or browser does not allow installing and trusting this private CA, camera access from that phone browser will not work. Use an organization-approved certificate/domain instead. Do not bypass certificate warnings or use plain HTTP as a camera-access workaround. When finished, remove the installed CA profile from the phone's certificate settings if you no longer need it; the private CA key remains on the PC. If you provide a custom TLS certificate with `python main.py --tls-cert CERT.pem --tls-key KEY.pem`, the app does not open its local CA download listener; configure your certificate's CA trust on the phone separately. + +## Windows Firewall and network limits + +Allow the app through Windows Defender Firewall on **Private networks only** when prompted. HTTPS signaling uses TCP port `8765`, the CA certificate download uses TCP port `8766`, and WebRTC media requires direct UDP connectivity. If creating rules manually, run PowerShell as Administrator and restrict access to the Private profile and local subnet: + +```powershell +$python = (Get-Command python).Source +New-NetFirewallRule -DisplayName "ADJU PHONE CAMERA signaling" -Direction Inbound -Program $python -Action Allow -Profile Private -Protocol TCP -LocalPort 8765 -RemoteAddress LocalSubnet +New-NetFirewallRule -DisplayName "ADJU PHONE CAMERA CA download" -Direction Inbound -Program $python -Action Allow -Profile Private -Protocol TCP -LocalPort 8766 -RemoteAddress LocalSubnet +New-NetFirewallRule -DisplayName "ADJU PHONE CAMERA WebRTC" -Direction Inbound -Program $python -Action Allow -Profile Private -Protocol UDP -RemoteAddress LocalSubnet +``` + +Replace the executable with the packaged app path when using the PyInstaller build, and adjust both TCP ports if you chose a different port. Remove these named rules when you no longer need them. This app deliberately has no TURN relay fallback. Do not disable the firewall globally. Guest Wi-Fi, client/AP isolation, VPN policy, or restrictive router/firewall rules may block the phone-to-PC connection. Both devices must be on a network that permits peer-to-peer LAN traffic. + +If the phone shows its camera preview but reports that it is still connecting, the camera permission and local capture have succeeded; the direct WebRTC video path is still negotiating. The PC shows **Phone paired** after PIN authentication and changes to **Receiving video** only after a frame arrives. If it stays at **Phone paired** or shows **WebRTC failed**, confirm the PC's Private-network UDP firewall rule, that the phone is not on guest Wi-Fi, and that both devices are still on the same local network. The phone's “answer received” status means signaling succeeded, not that video is already flowing. + +## Build a Windows executable + +From this directory with the virtual environment active: + +```powershell +python -m pip install pyinstaller +pyinstaller --noconfirm --clean --onefile --windowed --name "ADJU PHONE CAMERA" --add-data "adju_phone_camera\web;adju_phone_camera\web" main.py +``` + +The executable is created under `dist`. Windows Firewall may prompt for network access the first time it runs. The HTTPS certificate and key remain user-supplied files and are not embedded. + +## Tests + +```powershell +python -m pip install -r requirements-dev.txt +python -m pytest +``` + +The test suite covers PIN validation and throttling. End-to-end video additionally requires a supported phone browser, camera permission, a trusted HTTPS setup, and LAN connectivity. diff --git a/adju-phone-camera/adju_phone_camera/__init__.py b/adju-phone-camera/adju_phone_camera/__init__.py new file mode 100644 index 00000000..b3396d40 --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/__init__.py @@ -0,0 +1,3 @@ +"""ADJU PHONE CAMERA desktop application.""" + +__version__ = "0.1.0" diff --git a/adju-phone-camera/adju_phone_camera/__main__.py b/adju-phone-camera/adju_phone_camera/__main__.py new file mode 100644 index 00000000..5d6a8109 --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/__main__.py @@ -0,0 +1,3 @@ +from .main import main + +main() diff --git a/adju-phone-camera/adju_phone_camera/app.py b/adju-phone-camera/adju_phone_camera/app.py new file mode 100644 index 00000000..734b5e26 --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/app.py @@ -0,0 +1,300 @@ +"""CustomTkinter interface for ADJU PHONE CAMERA.""" + +from __future__ import annotations + +import threading +import time +from pathlib import Path +from tkinter import filedialog, messagebox + +import customtkinter as ctk +import qrcode +from PIL import Image, ImageTk + +from .pairing import generate_pairing_pin +from .server import CameraServer, get_lan_ip +from .certificates import create_local_certificate, default_certificate_directory + + +class CameraApp(ctk.CTk): + def __init__( + self, + port: int = 8765, + tls_cert: Path | None = None, + tls_key: Path | None = None, + ) -> None: + super().__init__() + self.title("ADJU PHONE CAMERA") + self.geometry("1080x760") + self.minsize(820, 620) + ctk.set_appearance_mode("dark") + ctk.set_default_color_theme("blue") + + self.port = port + self._managed_certificate = None + if tls_cert is None and tls_key is None: + self._managed_certificate = create_local_certificate(default_certificate_directory(), get_lan_ip()) + tls_cert = self._managed_certificate.certificate_path + tls_key = self._managed_certificate.private_key_path + self.tls_cert = tls_cert + self.tls_key = tls_key + self._ca_port = port + 1 if self._managed_certificate else None + self.pin = generate_pairing_pin() + self._image_lock = threading.Lock() + self._latest_frame: Image.Image | None = None + self._frame_revision = 0 + self._rendered_revision = -1 + self._frame_count = 0 + self._fps = 0 + self._fps_started = time.monotonic() + self._running = False + self._fullscreen = False + self._preview_photo: ImageTk.PhotoImage | None = None + self._server = CameraServer( + self.pin, + self._web_root(), + self._receive_frame, + self._set_status_from_server, + ca_download_path=( + self._managed_certificate.ca_download_path if self._managed_certificate else None + ), + ) + + self._build_ui() + self._refresh_access_link() + self.protocol("WM_DELETE_WINDOW", self._close) + self.bind("", lambda _event: self._toggle_fullscreen(False)) + self.after(30, self._update_preview) + + @staticmethod + def _web_root() -> Path: + import sys + + bundle_root = getattr(sys, "_MEIPASS", None) + if bundle_root: + return Path(bundle_root) / "adju_phone_camera" / "web" + return Path(__file__).resolve().parent / "web" + + def _build_ui(self) -> None: + self.grid_columnconfigure(0, weight=1) + self.grid_rowconfigure(2, weight=1) + + header = ctk.CTkFrame(self, corner_radius=14) + header.grid(row=0, column=0, padx=20, pady=(20, 12), sticky="ew") + header.grid_columnconfigure(0, weight=1) + ctk.CTkLabel(header, text="ADJU PHONE CAMERA", font=ctk.CTkFont(size=24, weight="bold")).grid( + row=0, column=0, padx=20, pady=(16, 2), sticky="w" + ) + ctk.CTkLabel( + header, + text="Use your phone as a camera over your local Wi-Fi — video stays on your network.", + text_color=("gray35", "gray75"), + ).grid(row=1, column=0, padx=20, pady=(0, 16), sticky="w") + self.status_label = ctk.CTkLabel(header, text="Stopped", text_color="#f0ad4e") + self.status_label.grid(row=0, column=1, rowspan=2, padx=20) + + controls = ctk.CTkFrame(self, corner_radius=14) + controls.grid(row=1, column=0, padx=20, pady=8, sticky="ew") + controls.grid_columnconfigure(1, weight=1) + self.start_button = ctk.CTkButton(controls, text="Start server", command=self._start_server) + self.start_button.grid(row=0, column=0, padx=(16, 10), pady=16) + self.stop_button = ctk.CTkButton( + controls, text="Stop server", command=self._stop_server, state="disabled", fg_color="#7d3434" + ) + self.stop_button.grid(row=0, column=1, padx=8, pady=16, sticky="w") + self.quality_menu = ctk.CTkOptionMenu( + controls, + values=["480p", "720p", "1080p"], + command=self._quality_changed, + ) + self.quality_menu.set("720p") + self.quality_menu.grid(row=0, column=2, padx=(8, 16), pady=16) + + body = ctk.CTkFrame(self, corner_radius=14) + body.grid(row=2, column=0, padx=20, pady=8, sticky="nsew") + body.grid_columnconfigure(0, weight=1) + body.grid_rowconfigure(0, weight=1) + + self.preview = ctk.CTkLabel( + body, + text="Start the server, scan the QR code, then pair and start the camera on your phone.", + justify="center", + font=ctk.CTkFont(size=16), + ) + self.preview.grid(row=0, column=0, padx=16, pady=16, sticky="nsew") + self.preview.bind("", lambda _event: self._toggle_fullscreen()) + + pairing = ctk.CTkFrame(body, width=250) + pairing.grid(row=0, column=1, padx=(0, 16), pady=16, sticky="ns") + pairing.grid_propagate(False) + ctk.CTkLabel(pairing, text="PAIR YOUR PHONE", font=ctk.CTkFont(weight="bold")).pack(pady=(18, 8)) + self.qr_label = ctk.CTkLabel(pairing, text="QR will appear here") + self.qr_label.pack(padx=16, pady=8) + self.pin_label = ctk.CTkLabel(pairing, text=f"PIN {self.pin}", font=ctk.CTkFont(size=22, weight="bold")) + self.pin_label.pack(pady=8) + self.url_label = ctk.CTkLabel(pairing, text="Server is stopped", wraplength=220, justify="center") + self.url_label.pack(padx=14, pady=8) + self.ca_url_label = ctk.CTkLabel(pairing, text="", wraplength=220, justify="center", font=ctk.CTkFont(size=10)) + self.ca_url_label.pack(padx=10, pady=(0, 3)) + self.ca_fingerprint_label = ctk.CTkLabel( + pairing, text="", wraplength=220, justify="center", font=ctk.CTkFont(size=9) + ) + self.ca_fingerprint_label.pack(padx=10, pady=(0, 5)) + self.fps_label = ctk.CTkLabel(pairing, text="0 FPS") + self.fps_label.pack(pady=(8, 2)) + ctk.CTkButton(pairing, text="Save screenshot", command=self._save_screenshot).pack( + padx=14, pady=8, fill="x" + ) + ctk.CTkButton(pairing, text="Fullscreen preview", command=self._toggle_fullscreen).pack( + padx=14, pady=(0, 16), fill="x" + ) + + ctk.CTkLabel( + self, + text="Before scanning: download and trust the CA from the link beside the QR, then scan the HTTPS QR and pair using the PIN.", + text_color=("#8b5a00", "#f0c36d"), + ).grid(row=3, column=0, padx=22, pady=(2, 14), sticky="w") + + def _refresh_access_link(self) -> None: + selected_resolution = self.quality_menu.get().removesuffix("p") + lan_ip = get_lan_ip() + self._url = f"https://{lan_ip}:{self.port}/?resolution={selected_resolution}" + self.url_label.configure(text=self._url) + qr = qrcode.make(self._url).convert("RGB") + self._qr_photo = ctk.CTkImage(light_image=qr, dark_image=qr, size=(150, 150)) + self.qr_label.configure(image=self._qr_photo, text="") + if self._managed_certificate: + self.ca_url_label.configure( + text=f"Before scanning, download the phone CA:\nhttp://{lan_ip}:{self._ca_port}/ca.cer" + ) + self.ca_fingerprint_label.configure( + text=f"Compare SHA-256 fingerprint:\n{self._managed_certificate.fingerprint}" + ) + else: + self.ca_url_label.configure( + text="Install and trust the CA for your HTTPS certificate on the phone first." + ) + self.ca_fingerprint_label.configure(text="") + + def _quality_changed(self, _value: str) -> None: + self._refresh_access_link() + + def _start_server(self) -> None: + if self._running: + return + try: + self._server.start( + "0.0.0.0", + self.port, + self.tls_cert, + self.tls_key, + ca_port=self._ca_port, + ) + except (OSError, ValueError) as error: + messagebox.showerror("Could not start server", str(error)) + return + self._running = True + self.start_button.configure(state="disabled") + self.stop_button.configure(state="normal") + self._set_status("Starting local server") + self.after(500, self._check_server_start) + + def _check_server_start(self) -> None: + if not self._running: + return + if not self._server.is_alive: + self._running = False + self.start_button.configure(state="normal") + self.stop_button.configure(state="disabled") + self._set_status("Server failed to start") + detail = self._server._startup_error or "Check whether the HTTPS and CA download ports are available." + messagebox.showerror( + "Could not start server", + f"The local server did not start.\n\n{detail}", + ) + return + if self._server.is_ready: + self._set_status("Waiting for phone") + return + self.after(250, self._check_server_start) + + def _stop_server(self) -> None: + if not self._running: + return + self._running = False + self._server.stop() + self.start_button.configure(state="normal") + self.stop_button.configure(state="disabled") + self._set_status("Stopped") + self.preview.configure(image="", text="Server stopped.") + self._preview_photo = None + with self._image_lock: + self._latest_frame = None + + def _set_status_from_server(self, status: str) -> None: + self.after(0, lambda: self._set_status(status)) + + def _set_status(self, status: str) -> None: + if not self.winfo_exists(): + return + self.status_label.configure(text=status) + + def _receive_frame(self, image: Image.Image) -> None: + with self._image_lock: + self._latest_frame = image + self._frame_revision += 1 + self._frame_count += 1 + elapsed = time.monotonic() - self._fps_started + if elapsed >= 1: + self._fps = round(self._frame_count / elapsed) + self._frame_count = 0 + self._fps_started = time.monotonic() + + def _update_preview(self) -> None: + if not self.winfo_exists(): + return + with self._image_lock: + frame = self._latest_frame + revision = self._frame_revision + fps = self._fps + if frame is not None and revision != self._rendered_revision: + preview_width = max(self.preview.winfo_width(), 320) + preview_height = max(self.preview.winfo_height(), 240) + image = frame.copy() + image.thumbnail((preview_width, preview_height), Image.Resampling.LANCZOS) + self._preview_photo = ImageTk.PhotoImage(image) + self.preview.configure(image=self._preview_photo, text="") + self._rendered_revision = revision + self.fps_label.configure(text=f"{fps} FPS") + self.after(30, self._update_preview) + + def _save_screenshot(self) -> None: + with self._image_lock: + frame = self._latest_frame.copy() if self._latest_frame is not None else None + if frame is None: + messagebox.showinfo("No video yet", "Pair a phone and start its camera before saving a screenshot.") + return + filename = filedialog.asksaveasfilename( + title="Save camera screenshot", + defaultextension=".png", + filetypes=[("PNG image", "*.png"), ("JPEG image", "*.jpg;*.jpeg")], + ) + if filename: + try: + frame.save(filename) + except OSError as error: + messagebox.showerror("Could not save screenshot", str(error)) + + def _toggle_fullscreen(self, enabled: bool | None = None) -> None: + self._fullscreen = not self._fullscreen if enabled is None else enabled + self.attributes("-fullscreen", self._fullscreen) + + def _close(self) -> None: + self._running = False + self._server.stop() + self.destroy() + + +def run_app(port: int, tls_cert: Path | None, tls_key: Path | None) -> None: + app = CameraApp(port=port, tls_cert=tls_cert, tls_key=tls_key) + app.mainloop() diff --git a/adju-phone-camera/adju_phone_camera/certificates.py b/adju-phone-camera/adju_phone_camera/certificates.py new file mode 100644 index 00000000..9de74937 --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/certificates.py @@ -0,0 +1,156 @@ +"""Generate and persist a private local certificate authority for LAN HTTPS.""" + +from __future__ import annotations + +import ipaddress +import os +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID + + +@dataclass(frozen=True) +class LocalCertificate: + certificate_path: Path + private_key_path: Path + ca_certificate_path: Path + ca_download_path: Path + fingerprint: str + + +def default_certificate_directory() -> Path: + local_app_data = os.environ.get("LOCALAPPDATA") + root = Path(local_app_data) if local_app_data else Path.home() / "AppData" / "Local" + return root / "ADJU PHONE CAMERA" / "certificates" + + +def _write_private_key(path: Path, key: ec.EllipticCurvePrivateKey) -> None: + path.write_bytes( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + ) + if os.name != "nt": + path.chmod(0o600) + + +def _load_or_create_ca(directory: Path) -> tuple[ec.EllipticCurvePrivateKey, x509.Certificate]: + key_path = directory / "local-ca-key.pem" + cert_path = directory / "local-ca-cert.pem" + if key_path.exists() != cert_path.exists(): + raise RuntimeError( + f"Local certificate files are incomplete in {directory}; keep or restore both CA files together." + ) + + if key_path.exists(): + key = serialization.load_pem_private_key(key_path.read_bytes(), password=None) + certificate = x509.load_pem_x509_certificate(cert_path.read_bytes()) + if not isinstance(key, ec.EllipticCurvePrivateKey): + raise RuntimeError("The saved local certificate authority key has an unsupported type.") + if key.public_key().public_numbers() != certificate.public_key().public_numbers(): + raise RuntimeError("The saved local certificate authority key does not match its certificate.") + return key, certificate + + key = ec.generate_private_key(ec.SECP256R1()) + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "ADJU PHONE CAMERA Local CA")]) + now = datetime.now(timezone.utc) + certificate = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(subject) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(minutes=5)) + .not_valid_after(now + timedelta(days=3650)) + .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) + .add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False) + .add_extension( + x509.AuthorityKeyIdentifier.from_issuer_public_key(key.public_key()), + critical=False, + ) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=True, + crl_sign=True, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .sign(key, hashes.SHA256()) + ) + _write_private_key(key_path, key) + cert_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM)) + return key, certificate + + +def create_local_certificate(directory: Path, lan_ip: str) -> LocalCertificate: + """Create/load a local CA and issue a TLS certificate for the current LAN IP.""" + address = ipaddress.ip_address(lan_ip) + if not isinstance(address, ipaddress.IPv4Address) or address.is_loopback or address.is_unspecified: + raise ValueError(f"A reachable private IPv4 address is required for phone HTTPS: {lan_ip}") + directory.mkdir(parents=True, exist_ok=True) + ca_key, ca_certificate = _load_or_create_ca(directory) + + server_key = ec.generate_private_key(ec.SECP256R1()) + now = datetime.now(timezone.utc) + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, str(address))]) + subject_alt_names = [ + x509.IPAddress(address), + x509.IPAddress(ipaddress.ip_address("127.0.0.1")), + x509.DNSName("localhost"), + ] + certificate = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(ca_certificate.subject) + .public_key(server_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(minutes=5)) + .not_valid_after(now + timedelta(days=90)) + .add_extension( + x509.SubjectAlternativeName(subject_alt_names), + critical=False, + ) + .add_extension( + x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), + critical=False, + ) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .add_extension(x509.SubjectKeyIdentifier.from_public_key(server_key.public_key()), critical=False) + .add_extension( + x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), + critical=False, + ) + .sign(ca_key, hashes.SHA256()) + ) + cert_path = directory / "server-cert.pem" + key_path = directory / "server-key.pem" + _write_private_key(key_path, server_key) + cert_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM)) + + ca_download = directory / "phone-trust.cer" + ca_download.write_bytes(ca_certificate.public_bytes(serialization.Encoding.DER)) + fingerprint = ca_certificate.fingerprint(hashes.SHA256()).hex().upper() + fingerprint = ":".join(fingerprint[index : index + 2] for index in range(0, len(fingerprint), 2)) + fingerprint = ca_certificate.fingerprint(hashes.SHA256()).hex().upper() + fingerprint = ":".join(fingerprint[index : index + 2] for index in range(0, len(fingerprint), 2)) + return LocalCertificate( + certificate_path=cert_path, + private_key_path=key_path, + ca_certificate_path=directory / "local-ca-cert.pem", + ca_download_path=ca_download, + fingerprint=fingerprint, + ) diff --git a/adju-phone-camera/adju_phone_camera/main.py b/adju-phone-camera/adju_phone_camera/main.py new file mode 100644 index 00000000..da1b58de --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/main.py @@ -0,0 +1,34 @@ +"""Command-line entry point.""" + +from __future__ import annotations + +import argparse +import logging +from pathlib import Path + +from .app import run_app + + +def main() -> None: + parser = argparse.ArgumentParser(description="Use an Android or iPhone camera as a local PC webcam preview.") + parser.add_argument("--port", type=int, default=8765, help="Local HTTPS and signaling port (default: 8765)") + parser.add_argument("--tls-cert", type=Path, help="PEM certificate for trusted local HTTPS") + parser.add_argument("--tls-key", type=Path, help="PEM private key matching --tls-cert") + parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) + args = parser.parse_args() + logging.basicConfig(level=getattr(logging, args.log_level), format="%(asctime)s %(levelname)s %(message)s") + + if not 1024 <= args.port <= 65535: + parser.error("--port must be between 1024 and 65535.") + if args.tls_cert is None and args.port == 65535: + parser.error("The default local CA download listener needs the port after --port; choose 65534 or lower.") + if (args.tls_cert is None) != (args.tls_key is None): + parser.error("--tls-cert and --tls-key must be supplied together.") + for label, path in (("--tls-cert", args.tls_cert), ("--tls-key", args.tls_key)): + if path is not None and not path.is_file(): + parser.error(f"{label} file does not exist: {path}") + run_app(args.port, args.tls_cert, args.tls_key) + + +if __name__ == "__main__": + main() diff --git a/adju-phone-camera/adju_phone_camera/pairing.py b/adju-phone-camera/adju_phone_camera/pairing.py new file mode 100644 index 00000000..fcfc866c --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/pairing.py @@ -0,0 +1,64 @@ +"""Pairing PIN helpers and per-address brute-force throttling.""" + +from __future__ import annotations + +import secrets +import time +from collections import defaultdict, deque + + +PIN_LENGTH = 6 + + +def generate_pairing_pin() -> str: + """Return a zero-padded, cryptographically random six-digit PIN.""" + return f"{secrets.randbelow(10**PIN_LENGTH):0{PIN_LENGTH}d}" + + +def is_valid_pin(pin: object) -> bool: + return isinstance(pin, str) and len(pin) == PIN_LENGTH and pin.isascii() and pin.isdigit() + + +class PinAttemptLimiter: + """Temporarily throttle repeated bad PIN attempts from one client address.""" + + def __init__( + self, + max_attempts: int = 5, + window_seconds: float = 60, + lockout_seconds: float = 60, + clock=time.monotonic, + ) -> None: + self.max_attempts = max_attempts + self.window_seconds = window_seconds + self.lockout_seconds = lockout_seconds + self._clock = clock + self._attempts: dict[str, deque[float]] = defaultdict(deque) + self._locked_until: dict[str, float] = {} + + def is_blocked(self, address: str) -> bool: + now = self._clock() + locked_until = self._locked_until.get(address, 0) + if now < locked_until: + return True + if locked_until: + self._locked_until.pop(address, None) + + attempts = self._attempts[address] + while attempts and now - attempts[0] >= self.window_seconds: + attempts.popleft() + return False + + def record_failure(self, address: str) -> None: + now = self._clock() + attempts = self._attempts[address] + while attempts and now - attempts[0] >= self.window_seconds: + attempts.popleft() + attempts.append(now) + if len(attempts) >= self.max_attempts: + self._locked_until[address] = now + self.lockout_seconds + attempts.clear() + + def record_success(self, address: str) -> None: + self._attempts.pop(address, None) + self._locked_until.pop(address, None) diff --git a/adju-phone-camera/adju_phone_camera/server.py b/adju-phone-camera/adju_phone_camera/server.py new file mode 100644 index 00000000..ea8f7c47 --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/server.py @@ -0,0 +1,305 @@ +"""Local HTTP/WebSocket signaling and direct WebRTC video receiving.""" + +from __future__ import annotations + +import asyncio +import hmac +import logging +import socket +import threading +from pathlib import Path +from typing import Callable + +import uvicorn +from aiortc import RTCPeerConnection, RTCSessionDescription, RTCConfiguration +from fastapi import FastAPI, WebSocket, WebSocketDisconnect +from fastapi.responses import FileResponse, PlainTextResponse +from fastapi.staticfiles import StaticFiles +from PIL import Image + +from .pairing import PinAttemptLimiter, is_valid_pin + +LOGGER = logging.getLogger(__name__) + + +def get_lan_ip() -> str: + """Find a private IPv4 address without sending packets to the internet.""" + try: + addresses = socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET) + except OSError: + addresses = [] + for address in addresses: + ip = address[4][0] + octets = ip.split(".") + if len(octets) != 4: + continue + first, second = int(octets[0]), int(octets[1]) + if first == 10 or (first == 172 and 16 <= second <= 31) or (first == 192 and second == 168): + return ip + return "127.0.0.1" + + +class CameraServer: + """Owns the local signaling server and one paired phone connection.""" + + def __init__( + self, + pin: str, + web_root: Path, + on_frame: Callable[[Image.Image], None], + on_status: Callable[[str], None], + ca_download_path: Path | None = None, + ) -> None: + if not is_valid_pin(pin): + raise ValueError("The pairing PIN must contain exactly six digits.") + self.pin = pin + self.web_root = web_root + self.on_frame = on_frame + self.on_status = on_status + self.ca_download_path = ca_download_path + self.app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None) + self._limiter = PinAttemptLimiter() + self._peer_connection: RTCPeerConnection | None = None + self._paired_socket: WebSocket | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._server: uvicorn.Server | None = None + self._servers: list[uvicorn.Server] = [] + self._startup_error: str | None = None + self.app.mount("/static", StaticFiles(directory=self.web_root), name="static") + self._register_routes() + + @property + def is_alive(self) -> bool: + return self._thread is not None and self._thread.is_alive() + + @property + def is_ready(self) -> bool: + return bool(self._servers) and all(server.started for server in self._servers) + + def _register_routes(self) -> None: + @self.app.get("/") + async def home() -> FileResponse: + return FileResponse(self.web_root / "index.html") + + @self.app.get("/health") + async def health() -> dict[str, str]: + return {"status": "ok"} + + @self.app.websocket("/ws") + async def signaling(websocket: WebSocket) -> None: + address = websocket.client.host if websocket.client else "unknown" + await websocket.accept() + paired = False + try: + while True: + message = await websocket.receive_json() + if not isinstance(message, dict): + await websocket.send_json({"type": "error", "message": "Invalid message."}) + continue + + message_type = message.get("type") + if not paired: + if message_type != "pair": + await websocket.send_json({"type": "error", "message": "Enter the pairing PIN first."}) + continue + if self._limiter.is_blocked(address): + await websocket.send_json( + {"type": "error", "message": "Too many incorrect PIN attempts. Wait one minute."} + ) + continue + supplied_pin = message.get("pin") + if not is_valid_pin(supplied_pin) or not hmac.compare_digest(self.pin, supplied_pin): + self._limiter.record_failure(address) + await websocket.send_json({"type": "error", "message": "Incorrect pairing PIN."}) + continue + self._limiter.record_success(address) + if self._paired_socket is not None and self._paired_socket is not websocket: + await websocket.send_json( + {"type": "error", "message": "Another phone is already paired."} + ) + continue + self._paired_socket = websocket + paired = True + self.on_status("Phone paired") + await websocket.send_json({"type": "paired"}) + continue + + if message_type == "offer": + await self._accept_offer(websocket, message) + elif message_type == "stop": + await self._close_peer() + self.on_status("Phone paired — camera stopped") + await websocket.send_json({"type": "stopped"}) + else: + await websocket.send_json({"type": "error", "message": "Unsupported signaling message."}) + except WebSocketDisconnect: + pass + except (ValueError, KeyError) as error: + LOGGER.info("Rejected signaling message from %s: %s", address, error) + try: + await websocket.send_json({"type": "error", "message": "Invalid WebRTC offer."}) + except RuntimeError: + pass + except Exception: + LOGGER.exception("Phone signaling connection failed") + try: + await websocket.send_json( + {"type": "error", "message": "Connection error. Reconnect and try again."} + ) + except RuntimeError: + pass + finally: + if self._paired_socket is websocket: + self._paired_socket = None + await self._close_peer() + self.on_status("Waiting for phone") + + async def _accept_offer(self, websocket: WebSocket, message: dict[str, object]) -> None: + sdp = message.get("sdp") + if not isinstance(sdp, str) or len(sdp) > 2_000_000: + raise ValueError("Invalid SDP.") + await self._close_peer() + peer = RTCPeerConnection(configuration=RTCConfiguration(iceServers=[])) + self._peer_connection = peer + frame_received = False + + @peer.on("track") + async def on_track(track) -> None: + nonlocal frame_received + if track.kind != "video": + return + try: + while True: + frame = await track.recv() + image = frame.to_image().convert("RGB") + if not frame_received: + frame_received = True + self.on_status("Receiving video") + self.on_frame(image) + except Exception: + LOGGER.info("Phone video track ended") + + @peer.on("connectionstatechange") + async def on_connection_state_change() -> None: + state = peer.connectionState + if state == "connected": + self.on_status("Receiving video" if frame_received else "WebRTC connected — waiting for video frames") + elif state in ("failed", "closed", "disconnected"): + self.on_status(f"WebRTC {state}") + + await peer.setRemoteDescription(RTCSessionDescription(sdp=sdp, type="offer")) + answer = await peer.createAnswer() + await peer.setLocalDescription(answer) + await websocket.send_json({"type": "answer", "sdp": peer.localDescription.sdp}) + + async def _close_peer(self) -> None: + peer = self._peer_connection + self._peer_connection = None + if peer is not None: + await peer.close() + + def _build_uvicorn_server( + self, + app: FastAPI, + host: str, + port: int, + tls_cert: Path | None = None, + tls_key: Path | None = None, + ) -> uvicorn.Server: + config = uvicorn.Config( + app, + host=host, + port=port, + log_level="warning", + log_config=None, + ssl_certfile=str(tls_cert) if tls_cert else None, + ssl_keyfile=str(tls_key) if tls_key else None, + ) + return uvicorn.Server(config) + + def start( + self, + host: str, + port: int, + tls_cert: Path | None = None, + tls_key: Path | None = None, + ca_port: int | None = None, + ) -> None: + if (tls_cert is None) != (tls_key is None): + raise ValueError("Both a TLS certificate and private key are required.") + if (self.ca_download_path is None) != (ca_port is None): + raise ValueError("A CA download certificate and HTTP port must be provided together.") + if self._thread is not None and self._thread.is_alive(): + return + self._startup_error = None + + server = self._build_uvicorn_server(self.app, host, port, tls_cert, tls_key) + self._server = server + self._servers = [server] + if ca_port is not None and self.ca_download_path is not None: + ca_app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None) + certificate_path = self.ca_download_path + + @ca_app.get("/ca.cer") + async def download_ca_certificate() -> FileResponse: + return FileResponse( + certificate_path, + media_type="application/pkix-cert", + filename="adju-phone-camera-local-ca.cer", + ) + + @ca_app.get("/") + async def ca_setup_info() -> PlainTextResponse: + return PlainTextResponse( + "ADJU PHONE CAMERA certificate download only. " + "Install the downloaded CA certificate only after comparing its fingerprint " + "with the desktop app. Pair and use the camera from the HTTPS QR code." + ) + + self._servers.append(self._build_uvicorn_server(ca_app, host, ca_port)) + + def serve() -> None: + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + + async def serve_all() -> None: + tasks = [asyncio.create_task(current.serve()) for current in self._servers] + try: + while not all(current.started for current in self._servers): + for task in tasks: + if task.done(): + task.result() + raise RuntimeError( + "An HTTP or HTTPS listener stopped before it was ready. " + "Check whether either port is already in use." + ) + await asyncio.sleep(0.05) + await asyncio.gather(*tasks) + finally: + for current in self._servers: + current.should_exit = True + await asyncio.gather(*tasks, return_exceptions=True) + + try: + self._loop.run_until_complete(serve_all()) + except Exception as error: + self._startup_error = str(error) + LOGGER.exception("Local camera HTTPS or certificate server failed") + self.on_status(f"Server error: {error}") + finally: + self._loop.close() + self._loop = None + + self._thread = threading.Thread(target=serve, name="adju-camera-server", daemon=True) + self._thread.start() + + def stop(self) -> None: + for server in self._servers: + server.should_exit = True + thread = self._thread + if thread is not None and thread.is_alive(): + thread.join(timeout=5) + self._thread = None + self._server = None + self._servers = [] diff --git a/adju-phone-camera/adju_phone_camera/web/app.js b/adju-phone-camera/adju_phone_camera/web/app.js new file mode 100644 index 00000000..8418f8ce --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/web/app.js @@ -0,0 +1,194 @@ +"use strict"; + +const statusElement = document.querySelector("#status"); +const pairForm = document.querySelector("#pair-form"); +const pinInput = document.querySelector("#pin"); +const pairButton = document.querySelector("#pair-button"); +const controls = document.querySelector("#camera-controls"); +const resolution = document.querySelector("#resolution"); +const facing = document.querySelector("#facing"); +const startButton = document.querySelector("#start-button"); +const stopButton = document.querySelector("#stop-button"); +const video = document.querySelector("#local-video"); + +let socket; +let peer; +let stream; +let paired = false; +let shouldReconnect = true; + +const requestedResolution = new URLSearchParams(location.search).get("resolution"); +if (["480", "720", "1080"].includes(requestedResolution)) { + resolution.value = requestedResolution; +} + +function setStatus(message) { + statusElement.textContent = message; +} + +function connect() { + const scheme = location.protocol === "https:" ? "wss:" : "ws:"; + socket = new WebSocket(`${scheme}//${location.host}/ws`); + socket.addEventListener("open", () => { + setStatus("Connected to PC. Enter the PIN shown in the desktop app."); + if (paired && pinInput.value) { + socket.send(JSON.stringify({ type: "pair", pin: pinInput.value })); + } + }); + socket.addEventListener("message", async (event) => { + try { + const message = JSON.parse(event.data); + if (message.type === "paired") { + paired = true; + pairForm.hidden = true; + controls.hidden = false; + startButton.disabled = false; + setStatus("Phone paired. Camera remains off until you start it."); + } else if (message.type === "answer" && peer) { + await peer.setRemoteDescription({ type: "answer", sdp: message.sdp }); + setStatus("Answer received. Connecting direct video to the PC…"); + } else if (message.type === "stopped") { + await closeCamera(false); + setStatus("Camera stopped. You can start it again when ready."); + } else if (message.type === "error") { + if (controls.hidden) { + paired = false; + pairButton.disabled = false; + } + setStatus(message.message || "The PC rejected the request."); + } + } catch (error) { + setStatus(`Invalid response or WebRTC negotiation error: ${error.message}`); + await closeCamera(true); + } + }); + socket.addEventListener("close", async () => { + await closeCamera(false); + setStatus("Connection lost. Reconnecting…"); + if (shouldReconnect) window.setTimeout(connect, 2000); + }); + socket.addEventListener("error", () => setStatus("Could not reach the PC. Check Wi-Fi and firewall settings.")); +} + +pairForm.addEventListener("submit", (event) => { + event.preventDefault(); + if (!socket || socket.readyState !== WebSocket.OPEN) { + setStatus("Not connected to the PC yet. Please wait and try again."); + return; + } + const pin = pinInput.value.trim(); + if (!/^[0-9]{6}$/.test(pin)) { + setStatus("Enter the six-digit PIN shown in the desktop app."); + return; + } + paired = true; + pairButton.disabled = true; + socket.send(JSON.stringify({ type: "pair", pin })); +}); + +startButton.addEventListener("click", async () => { + if (!window.isSecureContext || !navigator.mediaDevices?.getUserMedia) { + setStatus("Camera access requires HTTPS with a certificate trusted by this phone. See the desktop README."); + return; + } + startButton.disabled = true; + try { + const height = Number(resolution.value); + const width = Math.round(height * 16 / 9); + stream = await navigator.mediaDevices.getUserMedia({ + audio: false, + video: { + facingMode: { ideal: facing.value }, + width: { ideal: width }, + height: { ideal: height }, + }, + }); + video.srcObject = stream; + peer = new RTCPeerConnection({ iceServers: [] }); + for (const track of stream.getTracks()) peer.addTrack(track, stream); + peer.addEventListener("connectionstatechange", () => { + if (!peer) return; + if (peer.connectionState === "connected") { + setStatus("Direct video link connected. Waiting for frames at the PC…"); + } else if (peer.connectionState === "connecting") { + setStatus("Negotiating direct Wi-Fi video…"); + } else if (peer.connectionState === "disconnected") { + setStatus("Direct video link interrupted. Checking the connection…"); + } else if (peer.connectionState === "failed") { + setStatus("Direct video link failed. Check same Wi-Fi and allow UDP through the PC firewall."); + } else if (peer.connectionState === "closed") { + setStatus("Video connection closed."); + } + }); + const offer = await peer.createOffer(); + await peer.setLocalDescription(offer); + await waitForIceGathering(peer); + if (!socket || socket.readyState !== WebSocket.OPEN) throw new Error("The signaling connection was lost."); + socket.send(JSON.stringify({ type: "offer", sdp: peer.localDescription.sdp })); + stopButton.hidden = false; + resolution.disabled = true; + facing.disabled = true; + setStatus("Camera is on. Waiting for the PC to connect…"); + } catch (error) { + await closeCamera(false); + startButton.disabled = false; + if (error.name === "NotAllowedError" || error.name === "PermissionDeniedError") { + setStatus("Camera permission was denied. Allow camera access in the browser, then try again."); + } else if (error.name === "NotFoundError") { + setStatus("No camera was found on this phone."); + } else { + setStatus(error.message || "Could not start the camera."); + } + } +}); + +stopButton.addEventListener("click", async () => { + if (socket && socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ type: "stop" })); + await closeCamera(false); + setStatus("Camera stopped."); +}); + +async function closeCamera(enableStart) { + if (peer) { + peer.close(); + peer = null; + } + if (stream) { + for (const track of stream.getTracks()) track.stop(); + stream = null; + } + video.srcObject = null; + startButton.disabled = !paired || !enableStart; + stopButton.hidden = true; + resolution.disabled = false; + facing.disabled = false; +} + +function waitForIceGathering(connection) { + if (connection.iceGatheringState === "complete") return Promise.resolve(); + return new Promise((resolve) => { + let settled = false; + const finish = () => { + if (settled) return; + settled = true; + window.clearTimeout(timeout); + connection.removeEventListener("icegatheringstatechange", check); + resolve(); + }; + const check = () => { + if (connection.iceGatheringState === "complete") finish(); + }; + const timeout = window.setTimeout(finish, 10000); + connection.addEventListener("icegatheringstatechange", check); + check(); + }); +} + +window.addEventListener("beforeunload", () => { + shouldReconnect = false; + if (stream) for (const track of stream.getTracks()) track.stop(); + if (peer) peer.close(); + if (socket && socket.readyState === WebSocket.OPEN) socket.close(); +}); + +connect(); diff --git a/adju-phone-camera/adju_phone_camera/web/index.html b/adju-phone-camera/adju_phone_camera/web/index.html new file mode 100644 index 00000000..32f19638 --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/web/index.html @@ -0,0 +1,48 @@ + + + + + + + ADJU PHONE CAMERA + + + +
+
+

LOCAL CAMERA LINK

+

ADJU PHONE CAMERA

+

Pair this phone with your PC to stream video directly over Wi-Fi.

+
+ +
+

Connecting to your PC…

+
+ + + +
+ + +
+

Camera access starts only after you tap Start camera. Video is sent directly to your paired PC; this app does not use a cloud service or external video relay.

+
+ + + diff --git a/adju-phone-camera/adju_phone_camera/web/style.css b/adju-phone-camera/adju_phone_camera/web/style.css new file mode 100644 index 00000000..b85d75ec --- /dev/null +++ b/adju-phone-camera/adju_phone_camera/web/style.css @@ -0,0 +1,33 @@ +:root { + color-scheme: dark; + font-family: system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; + background: #10141d; + color: #eef2fa; +} + +* { box-sizing: border-box; } + +body { + margin: 0; + min-height: 100vh; + padding: 28px 16px; + background: radial-gradient(ellipse at top, #1c2a43, #10141d 62%); +} + +main { width: min(100%, 520px); margin: 0 auto; } +header { margin: 16px 4px 24px; } +h1 { margin: 4px 0 8px; font-size: clamp(1.7rem, 8vw, 2.4rem); letter-spacing: .03em; } +header p { color: #bac5d8; line-height: 1.5; } +.eyebrow { color: #83b4ff; font-size: .75rem; font-weight: 700; letter-spacing: .16em; } +.card { padding: 22px; border: 1px solid #29364b; border-radius: 18px; background: #171e2a; box-shadow: 0 18px 50px #0005; } +#status { min-height: 1.5em; color: #a9caff; font-weight: 600; } +label { display: block; margin: 18px 0 7px; color: #c5cede; font-size: .92rem; } +input, select, button { width: 100%; min-height: 48px; border: 1px solid #3a4960; border-radius: 10px; padding: 10px 12px; background: #101722; color: #fff; font: inherit; } +input { font-size: 1.3rem; letter-spacing: .35em; text-align: center; } +button { margin-top: 16px; border: 0; background: #3478e5; font-weight: 700; } +button:disabled { opacity: .5; } +button.secondary { margin-top: 10px; background: #39465b; } +.hint, .privacy { color: #a7b3c5; font-size: .85rem; line-height: 1.5; } +.privacy { margin: 18px 5px; } +video { display: block; width: 100%; max-height: 45vh; margin-top: 18px; border-radius: 12px; background: #080b10; object-fit: contain; } +[hidden] { display: none !important; } diff --git a/adju-phone-camera/main.py b/adju-phone-camera/main.py new file mode 100644 index 00000000..4f3a7620 --- /dev/null +++ b/adju-phone-camera/main.py @@ -0,0 +1,5 @@ +from adju_phone_camera.main import main + + +if __name__ == "__main__": + main() diff --git a/adju-phone-camera/requirements-dev.txt b/adju-phone-camera/requirements-dev.txt new file mode 100644 index 00000000..044d1338 --- /dev/null +++ b/adju-phone-camera/requirements-dev.txt @@ -0,0 +1,2 @@ +-r requirements.txt +pytest>=8,<9 diff --git a/adju-phone-camera/requirements.txt b/adju-phone-camera/requirements.txt new file mode 100644 index 00000000..c6ea1d48 --- /dev/null +++ b/adju-phone-camera/requirements.txt @@ -0,0 +1,7 @@ +aiortc>=1.9,<2 +customtkinter>=5.2,<6 +cryptography>=44,<51 +fastapi>=0.110,<1 +Pillow>=10,<13 +qrcode[pil]>=7.4,<9 +uvicorn[standard]>=0.29,<1 diff --git a/adju-phone-camera/tests/test_certificates.py b/adju-phone-camera/tests/test_certificates.py new file mode 100644 index 00000000..6b121774 --- /dev/null +++ b/adju-phone-camera/tests/test_certificates.py @@ -0,0 +1,52 @@ +import ipaddress +from pathlib import Path + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives import serialization + +from adju_phone_camera.certificates import create_local_certificate + + +def test_local_certificate_has_lan_ip_san_and_private_ca_download(tmp_path: Path): + material = create_local_certificate(tmp_path, "192.168.1.25") + certificate = x509.load_pem_x509_certificate(material.certificate_path.read_bytes()) + authority = x509.load_pem_x509_certificate(material.ca_certificate_path.read_bytes()) + downloaded_authority = x509.load_der_x509_certificate(material.ca_download_path.read_bytes()) + + san = certificate.extensions.get_extension_for_class(x509.SubjectAlternativeName).value + constraints = authority.extensions.get_extension_for_class(x509.BasicConstraints).value + private_key = serialization.load_pem_private_key(material.private_key_path.read_bytes(), password=None) + + assert san.get_values_for_type(x509.IPAddress) == [ + ipaddress.ip_address("192.168.1.25"), + ipaddress.ip_address("127.0.0.1"), + ] + assert san.get_values_for_type(x509.DNSName) == ["localhost"] + assert constraints.ca + assert downloaded_authority.fingerprint(hashes.SHA256()) == authority.fingerprint(hashes.SHA256()) + assert private_key.public_key().public_numbers() == certificate.public_key().public_numbers() + assert len(material.fingerprint.split(":")) == 32 + + +def test_existing_ca_is_reused_and_loopback_is_not_accepted(tmp_path: Path): + first = create_local_certificate(tmp_path, "10.0.0.12") + first_ca = first.ca_certificate_path.read_bytes() + second = create_local_certificate(tmp_path, "10.0.0.15") + second_server = x509.load_pem_x509_certificate(second.certificate_path.read_bytes()) + + assert second.ca_certificate_path.read_bytes() == first_ca + assert second.fingerprint == first.fingerprint + assert second_server.extensions.get_extension_for_class(x509.SubjectAlternativeName).value.get_values_for_type( + x509.IPAddress + )[0] == ipaddress.ip_address("10.0.0.15") + with pytest.raises(ValueError, match="private IPv4 address"): + create_local_certificate(tmp_path, "127.0.0.1") + + +def test_incomplete_ca_files_are_reported(tmp_path: Path): + (tmp_path / "local-ca-key.pem").write_bytes(b"partial") + + with pytest.raises(RuntimeError, match="incomplete"): + create_local_certificate(tmp_path, "192.168.1.10") diff --git a/adju-phone-camera/tests/test_pairing.py b/adju-phone-camera/tests/test_pairing.py new file mode 100644 index 00000000..60f62e9a --- /dev/null +++ b/adju-phone-camera/tests/test_pairing.py @@ -0,0 +1,39 @@ +from adju_phone_camera.pairing import PinAttemptLimiter, generate_pairing_pin, is_valid_pin + + +def test_generated_pin_is_six_ascii_digits(): + pin = generate_pairing_pin() + + assert len(pin) == 6 + assert pin.isascii() + assert pin.isdigit() + + +def test_pin_validation_rejects_non_six_digit_values(): + assert is_valid_pin("000001") + assert not is_valid_pin("12345") + assert not is_valid_pin("1234567") + assert not is_valid_pin("１２３４５６") + assert not is_valid_pin(None) + + +def test_limiter_blocks_after_threshold_and_expires(): + now = [0.0] + limiter = PinAttemptLimiter(max_attempts=2, window_seconds=30, lockout_seconds=10, clock=lambda: now[0]) + + assert not limiter.is_blocked("192.168.1.4") + limiter.record_failure("192.168.1.4") + limiter.record_failure("192.168.1.4") + assert limiter.is_blocked("192.168.1.4") + now[0] = 10.1 + assert not limiter.is_blocked("192.168.1.4") + + +def test_success_clears_previous_failures(): + now = [0.0] + limiter = PinAttemptLimiter(max_attempts=2, clock=lambda: now[0]) + limiter.record_failure("phone") + limiter.record_success("phone") + limiter.record_failure("phone") + + assert not limiter.is_blocked("phone") diff --git a/adju-phone-camera/tests/test_server.py b/adju-phone-camera/tests/test_server.py new file mode 100644 index 00000000..ccf84dfe --- /dev/null +++ b/adju-phone-camera/tests/test_server.py @@ -0,0 +1,134 @@ +import asyncio +import ssl +import time +import urllib.error +import urllib.request +from pathlib import Path +import socket +from types import SimpleNamespace + +from fastapi import WebSocketDisconnect +import pytest + +from adju_phone_camera.certificates import create_local_certificate +from adju_phone_camera.server import CameraServer + + +WEB_ROOT = Path(__file__).parents[1] / "adju_phone_camera" / "web" + + +def make_server(): + return CameraServer("123456", WEB_ROOT, lambda _image: None, lambda _status: None) + + +class FakeWebSocket: + def __init__(self, messages): + self.client = SimpleNamespace(host="192.168.1.10") + self.messages = iter(messages) + self.responses = [] + + async def accept(self): + pass + + async def receive_json(self): + try: + return next(self.messages) + except StopIteration: + raise WebSocketDisconnect(code=1000) + + async def send_json(self, message): + self.responses.append(message) + + +def test_phone_page_and_stylesheet_are_served(): + app = make_server().app + home_route = next(route for route in app.routes if getattr(route, "path", None) == "/") + page = asyncio.run(home_route.endpoint()) + static_mount = next(route for route in app.routes if getattr(route, "path", None) == "/static") + + assert Path(page.path).read_text(encoding="utf-8").find("ADJU PHONE CAMERA") >= 0 + assert (WEB_ROOT / "style.css").is_file() + assert static_mount.name == "static" + + +def test_signaling_requires_pin_and_can_stop_after_pairing(): + app = make_server().app + endpoint = next(route.endpoint for route in app.routes if getattr(route, "path", None) == "/ws") + websocket = FakeWebSocket( + [ + {"type": "offer", "sdp": "ignored"}, + {"type": "pair", "pin": "123456"}, + {"type": "stop"}, + ] + ) + + asyncio.run(endpoint(websocket)) + + assert [response["type"] for response in websocket.responses] == ["error", "paired", "stopped"] + + +def test_server_disables_uvicorn_logging_config_for_frozen_app(monkeypatch): + observed = {} + + class FakeServer: + def __init__(self, config): + observed["log_config"] = config.log_config + self.should_exit = False + + async def serve(self): + pass + + monkeypatch.setattr("adju_phone_camera.server.uvicorn.Server", FakeServer) + server = make_server() + server.start("127.0.0.1", 8765) + server._thread.join(timeout=2) + + assert not server._thread.is_alive() + assert observed["log_config"] is None + + +def _free_port() -> int: + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + return listener.getsockname()[1] + + +def test_https_and_http_ca_download_listeners(tmp_path): + material = create_local_certificate(tmp_path, "192.168.1.10") + https_port = _free_port() + ca_port = _free_port() + while ca_port == https_port: + ca_port = _free_port() + server = CameraServer( + "123456", + WEB_ROOT, + lambda _image: None, + lambda _status: None, + ca_download_path=material.ca_download_path, + ) + server.start( + "127.0.0.1", + https_port, + material.certificate_path, + material.private_key_path, + ca_port=ca_port, + ) + try: + deadline = time.monotonic() + 5 + while not server.is_ready and server.is_alive and time.monotonic() < deadline: + time.sleep(0.05) + assert server.is_ready, server._startup_error + + with urllib.request.urlopen(f"http://127.0.0.1:{ca_port}/ca.cer", timeout=2) as response: + ca_download = response.read() + assert ca_download == material.ca_download_path.read_bytes() + with pytest.raises(urllib.error.HTTPError) as response: + urllib.request.urlopen(f"http://127.0.0.1:{ca_port}/ws", timeout=2) + assert response.value.code == 404 + + context = ssl.create_default_context(cafile=str(material.ca_certificate_path)) + with urllib.request.urlopen(f"https://127.0.0.1:{https_port}/", context=context, timeout=2) as response: + assert response.status == 200 + assert "ADJU PHONE CAMERA" in response.read().decode("utf-8") + finally: + server.stop()