#!/usr/bin/env python3 """ PhotoDrop — hardened / clinical variant. Self-contained LAN photo drop that is offline and access-controlled: - UI is fully INLINE — hand-written CSS, emoji icons, system fonts. NO CDN, so it renders with **no internet** (the plain variant loaded Tailwind/Lucide/fonts from CDNs). - QR code is generated **locally** with `segno` — no api.qrserver.com, so the LAN URL never leaves the network. - A per-session **access token** (ridden inside the QR) gates every request, so only a phone that scanned THIS screen's QR can view the gallery or upload. Set PHOTODROP_TOKEN to pin your own. Requires: `pip install segno` (one small pure-python dependency, for offline QR). NOTE: still plain HTTP — the token blocks other devices on the Wi-Fi but is NOT encryption. For a hardened clinical deployment add TLS (see the recipe's "Encryption" section). """ import argparse import http.server import json import os import secrets import shutil import socket import socketserver import ssl import subprocess import sys import time import urllib.parse import webbrowser import segno # pip install segno # --- CONFIG --- PORT = 8000 UPLOAD_DIR = "dropped_photos" # Per-session access token (rides in the QR). Override with PHOTODROP_TOKEN for a fixed value. TOKEN = os.environ.get("PHOTODROP_TOKEN") or secrets.token_urlsafe(6) SCHEME = "http" # becomes "https" with --https, so the QR's embedded URL matches HTML_TEMPLATE = """ PhotoDrop LAN
""" DENIED_HTML = """PhotoDrop
🔒

Access denied

Open PhotoDrop from the host screen, or scan the QR code shown there.

""" def get_local_ip(): """Find the LAN IP via the standard UDP-route trick (no packet is sent).""" s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] except Exception: ip = "127.0.0.1" finally: s.close() return ip def _token_ok(path): q = urllib.parse.parse_qs(urllib.parse.urlparse(path).query) return secrets.compare_digest((q.get("token") or [""])[0], TOKEN) class Handler(http.server.SimpleHTTPRequestHandler): def _send(self, code, body, ctype="text/html"): b = body if isinstance(body, bytes) else body.encode("utf-8") self.send_response(code) self.send_header("Content-type", ctype) self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) def do_GET(self): p = urllib.parse.urlparse(self.path).path if p == "/favicon.ico": # Served ahead of the token gate: browsers request the favicon automatically # with no token, and a 403 here just litters the console. Inline emoji SVG. svg = ("" "\U0001F4F6") self._send(200, svg, "image/svg+xml") return if not _token_ok(self.path): self._send(403, DENIED_HTML) return if p == "/api/list": files = [] if os.path.isdir(UPLOAD_DIR): for f in os.listdir(UPLOAD_DIR): if not f.startswith("."): fp = os.path.join(UPLOAD_DIR, f) files.append({"name": f, "url": f"/{UPLOAD_DIR}/{f}", "timestamp": os.path.getmtime(fp)}) self._send(200, json.dumps(files), "application/json") return if p.startswith(f"/{UPLOAD_DIR}/"): self.path = p # drop the ?token query so the file handler serves the file super().do_GET() return # Frontend page — inject a locally-generated QR (offline) + the upload dir. # PHOTODROP_HOST overrides the auto-detected IP so a containerized / reverse-proxied # instance can advertise the host's LAN IP (get_local_ip() would return the container IP). ip = os.environ.get("PHOTODROP_HOST") or get_local_ip() mobile_url = f"{SCHEME}://{ip}:{PORT}/?mode=mobile&token={TOKEN}" qr_uri = segno.make(mobile_url, error="m").png_data_uri(scale=6, border=2) html = HTML_TEMPLATE.replace("{{QR_IMG}}", qr_uri).replace("{{UPLOAD_DIR}}", UPLOAD_DIR) self._send(200, html) def do_POST(self): if not _token_ok(self.path): self._send(403, "denied") return if urllib.parse.urlparse(self.path).path == "/api/upload": try: q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query) name = os.path.basename((q.get("name") or ["unknown.png"])[0]) n = int(self.headers["Content-Length"]) data = self.rfile.read(n) os.makedirs(UPLOAD_DIR, exist_ok=True) fp = os.path.join(UPLOAD_DIR, name) if os.path.exists(fp): base, ext = os.path.splitext(name) fp = os.path.join(UPLOAD_DIR, f"{base}_{int(time.time())}{ext}") with open(fp, "wb") as f: f.write(data) self._send(200, "Success") print(f"Received: {name}") except Exception as e: print("Error:", e) self._send(500, "error") def _gen_cert_cryptography(cert_path, key_path, ip): """Generate a self-signed cert with the `cryptography` lib (cross-platform, no openssl needed).""" import ipaddress from datetime import datetime, timedelta, timezone from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.x509.oid import NameOID key = rsa.generate_private_key(public_exponent=65537, key_size=2048) name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, ip)]) alt = [x509.DNSName("localhost")] try: alt.append(x509.IPAddress(ipaddress.ip_address(ip))) except ValueError: pass cert = ( x509.CertificateBuilder() .subject_name(name).issuer_name(name) .public_key(key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.now(timezone.utc) - timedelta(days=1)) .not_valid_after(datetime.now(timezone.utc) + timedelta(days=825)) .add_extension(x509.SubjectAlternativeName(alt), critical=False) .sign(key, hashes.SHA256()) ) with open(key_path, "wb") as f: f.write(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption())) with open(cert_path, "wb") as f: f.write(cert.public_bytes(serialization.Encoding.PEM)) def _gen_cert_openssl(cert_path, key_path, ip): subprocess.run( ["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", key_path, "-out", cert_path, "-days", "825", "-subj", f"/CN={ip}", "-addext", f"subjectAltName=IP:{ip},DNS:localhost"], check=True, ) def ensure_cert(cert_path, key_path, ip): """Make a self-signed cert if one isn't present yet (via cryptography, else openssl).""" if os.path.exists(cert_path) and os.path.exists(key_path): print(f"[tls] using existing cert: {cert_path}") return print(f"[tls] generating self-signed cert for {ip} -> {cert_path}") try: _gen_cert_cryptography(cert_path, key_path, ip) return except ImportError: pass if shutil.which("openssl"): _gen_cert_openssl(cert_path, key_path, ip) return sys.exit( "[tls] can't generate a cert. Do ONE of:\n" " - pip install cryptography (then re-run)\n" " - install openssl (then re-run)\n" " - bring your own: --cert cert.pem --key key.pem\n" f" manual: openssl req -x509 -newkey rsa:2048 -nodes -keyout {key_path} " f"-out {cert_path} -days 825 -subj /CN={ip}" ) def run(): global SCHEME, PORT ap = argparse.ArgumentParser( description="PhotoDrop (hardened): offline LAN photo drop, access-token gated.") ap.add_argument("--https", action="store_true", help="serve over TLS with a self-signed cert (auto-generated on first run)") ap.add_argument("--port", type=int, default=PORT, help=f"port (default {PORT})") ap.add_argument("--cert", default="cert.pem", help="TLS cert path (with --https)") ap.add_argument("--key", default="key.pem", help="TLS key path (with --https)") args = ap.parse_args() PORT = args.port os.makedirs(UPLOAD_DIR, exist_ok=True) ip = os.environ.get("PHOTODROP_HOST") or get_local_ip() httpd = socketserver.TCPServer(("0.0.0.0", PORT), Handler) if args.https: ensure_cert(args.cert, args.key, ip) ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) ctx.load_cert_chain(args.cert, args.key) httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True) SCHEME = "https" url = f"{SCHEME}://{ip}:{PORT}/?token={TOKEN}" tls_tag = " + TLS" if args.https else "" print("\n" + "=" * 52) print(f" PHOTODROP (hardened: offline + token{tls_tag}) STARTED") print("=" * 52) print(f" Desktop URL : {url}") print(" (open this — the token gates ALL access; the QR carries it to phones)") if args.https: print(" (self-signed TLS: your phone shows a one-time 'not private' warning — tap through)") print(f" Photos : {os.path.abspath(UPLOAD_DIR)}") print("=" * 52 + "\n") webbrowser.open(url) try: httpd.serve_forever() except KeyboardInterrupt: print("\nStopping server...") httpd.server_close() if __name__ == "__main__": run()