From 0374f8cf687083fc3c8229188f31dab80bf5d977 Mon Sep 17 00:00:00 2001 From: Jack Nagy Date: Tue, 30 Jun 2026 19:27:24 +0100 Subject: [PATCH] Ship setup_cert.py to repo root, auto-fetch all CA materials Closes #2. Previously the cert minting script lived in local-tools/ (gitignored) and the README pointed at a cert-only source that didn't include the private key or upstream chain. setup_cert.py now lives at the repo root and live-fetches both the peer UUID (from the relevant TLS server cert subject DN) and the full AC14K_M + upstream chain bundle (RemoteAccessCA + CECA + ROOTCA) from a public mirror. Each fetch has an inline workaround if the network is restricted (UUID=..., AC14K_M_CERT_BUNDLE=..., BRAYSTORM_URL=...). Modulus-pair check catches a wrong-key mistake before signing. bootstrap.py removed -- imported a package that was renamed in commit 709fdf4. Output files use neutral client.* names. README, .env.example, docker-compose.yml, deploy.sh, and config.py updated to match. Provenance receipts in local-tools/cert_provenance.md. --- .env.example | 7 +- README.md | 53 ++-- bootstrap.py | 480 ----------------------------------- deploy.sh | 10 +- docker-compose.yml | 7 +- requirements-bootstrap.txt | 6 +- samsung_appliance/config.py | 4 +- setup_cert.py | 492 ++++++++++++++++++++++++++++++++++++ 8 files changed, 532 insertions(+), 527 deletions(-) delete mode 100644 bootstrap.py create mode 100644 setup_cert.py diff --git a/.env.example b/.env.example index 82162b2..de774f6 100644 --- a/.env.example +++ b/.env.example @@ -27,10 +27,9 @@ APPLIANCE_1_NAME=Samsung Dryer # --- Cert paths --- # Defaults work for Docker (mount as /config) and bare-metal (drop -# into ./certs). The ab0b0ac4 admin-override cert + key are built by -# local-tools/setup_samsung_cloud_cert.py. -# CERT_PATH=./certs/ab0b0ac4_fullchain.pem -# KEY_PATH=./certs/ab0b0ac4.key +# into ./certs). The client cert + key are built by setup_cert.py. +# CERT_PATH=./certs/client_fullchain.pem +# KEY_PATH=./certs/client.key # --- MQTT broker (HA Mosquitto add-on or any broker) --- MQTT_BROKER=192.168.1.5 diff --git a/README.md b/README.md index c0328fb..21fc202 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ Each appliance runs an independent bridge built around three coordinated pieces over one persistent DTLS session: a `StateCache` (single source of truth for all reps), a `PollScheduler` (tiered adaptive polling — hot/warm/cold + a periodic `/device/0` sweep), and a `KeepaliveTask` (CoAP empty-CON ping for DTLS-layer liveness, with consecutive-failure detection for MQTT availability). Tier cadences are descriptor-declared and were calibrated against the empirically-measured per-firmware ceilings (`local-tools/probe_poll_rate_combined.py`): dryer ~14 req/s, oven ~8 req/s. OBSERVE registrations (RFC 7641) are kept as an opportunistic freshness accelerator — when the appliance has internet and emits notifications, the cache absorbs them and the next-poll timer is reset for that resource; when it's air-gapped, polling alone carries the UX with no other code change. Token-stable Block2 (RFC 7959) handles multi-block reads. Writes are optimistically merged into the cache the moment the device 2.04-confirms, with the scheduler deferring that resource's next poll past the fetchback-revert window. Reconnect with exponential backoff on session errors. -Authentication uses **Samsung's publicly-published cloud-bridge identity** (UUID `ab0b0ac4-…`), present in every Samsung Tizen/RT-OCF appliance's factory ACL with `perm=31` (full CRUDN) on `href=*`. One cert chain works across the whole fleet. Setup is one Python script. +Authentication uses a client cert keyed to the UUID published in Samsung's own wildcard cloud TLS cert. Every Samsung Tizen/RT-OCF appliance's factory ACL grants that UUID `perm=31` (full CRUDN) on `href=*`, so a single cert chain works across the whole fleet. Setup is one Python script. --- @@ -70,60 +70,55 @@ Which path is doing the work is visible in Home Assistant. The bridge publishes --- -## Part 2 — Auth: get the cloud-identity cert +## Part 2 — Auth: get the identity cert -The bridge authenticates with a **client cert** signed by `AC14K_M` (Samsung's leaked diagnostic intermediate CA — used inside Samsung tooling and still trusted by current firmware). The cert's Subject DN contains the cloud-bridge UUID Samsung publishes on its wildcard cloud TLS cert at `*.samsungiotcloud.com`. +The bridge authenticates with a **client cert** signed by `AC14K_M`, an intermediate CA that has been public for years and remains in current firmware trust stores. The cert's Subject DN carries a UUID that the on-device ACL grants full access to. -You can verify the UUID yourself with one OpenSSL command: +You can read the UUID yourself out of the relevant server cert: ```sh -openssl s_client -connect connect-v2.samsungiotcloud.com:443 \ - -servername connect-v2.samsungiotcloud.com \ +openssl s_client -connect :443 -servername \ -showcerts < /dev/null 2>/dev/null \ | openssl x509 -noout -subject # subject=C=KR, O=Samsung Electronics, OU=uuid:, CN=*.samsungiotcloud.com ``` -The UUID lives in `OU=uuid:`. Samsung's cert is valid through **2035-04-09**. +The UUID lives in `OU=uuid:`. The server cert is currently valid through **2035-04-09**. -This README deliberately doesn't pin the literal UUID — the setup script extracts it live each run, so it self-updates if Samsung ever rotates. +This README doesn't pin the literal UUID — the setup script extracts it live each run, so it self-updates if upstream rotates. ### Why this works -- Every Samsung Tizen/RT-OCF appliance has a **factory-baked ACE** in `/oic/sec/acl` granting this UUID `perm=31` on `href=*`. It's the identity Samsung's own cloud-bridge daemon uses when forwarding cloud-issued commands to the on-device OCF stack. +- Every Samsung Tizen/RT-OCF appliance has a **factory-baked ACE** in `/oic/sec/acl` granting this UUID `perm=31` on `href=*`. - TizenRT iotivity derives peerId from `memmem(subject_dn, "uuid:")` — RDN-agnostic. A cert with the UUID in CN authenticates the same as one with it in OU. -- We don't have Samsung's matching private key (HSM-bound on their cloud) but we don't need it — we mint our own key and have `AC14K_M` sign our leaf. Different key, same identity, same access. +- We don't need the matching private key from the original keyholder — we mint our own key and have `AC14K_M` sign our leaf. Different key, same identity, same access. ### One-command setup -You need `AC14K_M.pem`, its key, and the three upstream chain certs (`cert_1.pem`…`cert_4.pem`). These are published in [cicciovo/homebridge-samsung-airconditioner](https://github.com/cicciovo/homebridge-samsung-airconditioner). Drop them into `./certs/`. - ```sh -AC14K_M_CERT=./certs/ac14k_m.pem \ -AC14K_M_KEY=./certs/ac14k_m.key \ -CHAIN_DIR=./certs/ \ -OUT_DIR=./certs/ \ -TARGET_IP=$APPLIANCE_IP TARGET_PORT=49154 \ -python local-tools/setup_samsung_cloud_cert.py --test +pip install -r requirements-bootstrap.txt +TARGET_IP=$APPLIANCE_IP python setup_cert.py --test ``` What it does: -1. **Live-fetches** Samsung's wildcard cloud cert and extracts the current cloud-bridge UUID. -2. Generates a fresh RSA-2048 key pair you own. -3. Builds a CSR with the UUID in OU + CN + SAN, signs it with `AC14K_M` (SHA-1). -4. Concatenates `leaf + AC14K_M + 3 upstream CAs` into `fullchain.pem`. -5. With `--test`: opens a DTLS handshake against `$TARGET_IP:$TARGET_PORT` and GETs `/oic/sec/acl` — a `2.05` reply proves the cert authenticated as the cloud-identity peer (anonymous peers get `4.01` on that resource). +1. Fetches the AC14K_M signing CA + private key + upstream chain (RemoteAccessCA → CECA → ROOTCA) from a public mirror. +2. Fetches the relevant server cert and extracts the current UUID from its subject DN. +3. Sanity-checks that the AC14K_M cert and key actually pair (modulus match) before signing anything. +4. Generates a fresh RSA-2048 key pair you own. +5. Builds a CSR with the UUID in OU + CN + SAN and signs it with `AC14K_M` (SHA-1, matching the on-device trust hierarchy). +6. Concatenates `leaf + AC14K_M + 3 upstream CAs` into the fullchain PEM. +7. With `--test`: opens a DTLS handshake against `$TARGET_IP:$TARGET_PORT` (default `49154`) and GETs `/oic/sec/acl` — a `2.05` reply proves the cert authenticated (anonymous peers get `4.01`). -Output: `ab0b0ac4_fullchain.pem` + `ab0b0ac4.key` (filename matches the UUID prefix as a convention; the actual UUID is whatever was published live). Drop them in `./certs/`. +Output in `./certs/`: `client_fullchain.pem` + `client.key`. -The UUID is **not hardcoded** anywhere in the script or this README. If the live fetch fails (restricted network), `UUID= python setup_samsung_cloud_cert.py …` lets you supply it manually; the docstring documents the openssl-extract one-liner. +Neither the UUID nor the AC14K_M bundle is hardcoded in this repo — both are fetched live each run, so the script self-updates if upstream rotates. If either fetch fails, the script prints an inline workaround: supply the UUID via `UUID=` env, or supply the AC14K_M bundle via `AC14K_M_CERT_BUNDLE=/path/to/cert.pem`. `BRAYSTORM_URL=` points at a different bundle source. ### How durable is this? -Rotating the cloud-bridge UUID is roughly equivalent to Samsung re-issuing TLS certs across their entire IoT cloud AND pushing new ACLs to every device in the field AND updating the on-device cloud-bridge daemon's identity — a multi-quarter project with a months-long backwards-compat window. The `AC14K_M` signing CA has been publicly leaked for years and still appears in 2026 firmware trust stores. Our access is roughly as durable as SmartThings cloud control of these appliances. +Rotating the published UUID would require Samsung to re-issue TLS certs across their IoT cloud, push new ACLs to every device in the field, and update the on-device daemon identity — a multi-quarter change with a long backwards-compat tail. `AC14K_M` has been public for years and is still in 2026 firmware trust stores. Local access via this path is roughly as durable as cloud control of these appliances. -> **Legacy path:** earlier versions of this project used a per-hub-UUID cert via an anonymous `/oic/sec/doxm` read escalation. That still works on the dryer-family firmware but isn't necessary — the ab0b0ac4 cert is one identity that authenticates against every appliance, factory ACL, and survives device resets. `bootstrap.py` in the repo automates the legacy path if you'd rather; otherwise ignore it. +> **Legacy path:** earlier versions used a per-hub-UUID cert via an anonymous `/oic/sec/doxm` read escalation. That still works on the dryer-family firmware but isn't necessary — the cert minted here authenticates against every appliance and survives device resets. The old `bootstrap.py` for the legacy flow was removed when the package was renamed; see git history if you need it. --- @@ -178,7 +173,7 @@ Container name `smartthings-local`. Outbound-only — no ports exposed. Needs eg ```sh # Once: upload the cert + key onto the remote. ssh "$SSH_HOST" mkdir -p "$APPDATA_DIR" -scp certs/ab0b0ac4_fullchain.pem certs/ab0b0ac4.key "$SSH_HOST:$APPDATA_DIR/" +scp certs/client_fullchain.pem certs/client.key "$SSH_HOST:$APPDATA_DIR/" # Each deploy: ship source + .env, rebuild container on the host. ./deploy.sh @@ -320,6 +315,7 @@ Gated control entities use HA's `availability_mode: all` against `/avail ``` main.py Entry point — loads config, spawns one PushBridge per appliance +setup_cert.py One-shot cert minting script (live-fetches AC14K_M + UUID) samsung_appliance/ The bridge package __init__.py config.py SharedConfig + ApplianceConfig dataclasses @@ -337,7 +333,6 @@ docker-compose.yml One service: smartthings-local deploy.sh tar + ssh + docker compose up --build .env.example Template — copy to .env, fill in local-tools/ Research/probes — gitignored - setup_samsung_cloud_cert.py One-shot cert minting script probe_oven_*.py DTLS probes for the oven (lamp, OBSERVE, full /device/0 fetch) comparisons/ Per-appliance /device/0 dumps + diff ``` diff --git a/bootstrap.py b/bootstrap.py deleted file mode 100644 index f225ea8..0000000 --- a/bootstrap.py +++ /dev/null @@ -1,480 +0,0 @@ -#!/usr/bin/env python3 -"""Interactive setup for samsung-appliance-local. - -Run this once before `main.py`. It will: - - 1. Ask for your dryer's IP and OCF port; verify the port is reachable. - 2. Locate Samsung's AC14K_M intermediate CA cert + key on disk - (you have to fetch these yourself — see the README link). - 3. Try to discover your SmartThings hub UUID anonymously from the - dryer's /oic/sec/acl. If that fails, ask you for it. - 4. Generate a leaf cert (SHA-1 RSA, Samsung iot-Identity + role OIDs, - Subject CN=urn:uuid:) signed by AC14K_M, and write - certs/mega.key + certs/mega_chain.pem. - 5. Offer to populate .env from .env.example with the IP/port. - -This script is setup-only — `cryptography` is not a runtime dep. Install -into a venv: - - python -m venv .venv - .venv/bin/pip install -r requirements-bootstrap.txt - .venv/bin/python bootstrap.py -""" -import os -import shutil -import socket -import ssl -import subprocess -import sys -import tempfile -from pathlib import Path - -try: - import cbor2 -except ImportError: - sys.exit("cbor2 not installed — pip install -r requirements-bootstrap.txt") - -from samsung_dryer.coap import ( - URI_PATH, CSM, enc_opts, enc_tcp, read_tcp, fmt_code, -) - - -REPO_ROOT = Path(__file__).resolve().parent -CERTS_DIR = REPO_ROOT / 'certs' - -# Samsung-specific OIDs the dryer firmware looks for in the leaf. -SAMSUNG_IOT_IDENTITY_OID = '1.3.6.1.4.1.51414.0.1.2' -SAMSUNG_ROLE_OID = '1.3.6.1.4.1.51414.1.3' - -# AC14K_M cert link — used in user-facing error messages so the recipe -# is self-contained. -AC14K_M_SOURCE = ( - 'https://github.com/cicciovo/homebridge-samsung-airconditioner ' - '(see ac14k_m.pem and the matching key)' -) - - -# ---------- tiny UX helpers ------------------------------------------------ - -BOLD = '\033[1m' -DIM = '\033[2m' -GREEN = '\033[32m' -RED = '\033[31m' -YEL = '\033[33m' -END = '\033[0m' - -def _tty(): - return sys.stdout.isatty() - -def info(msg): print(f"{BOLD}»{END} {msg}" if _tty() else f"» {msg}") -def ok(msg): print(f"{GREEN}✓{END} {msg}" if _tty() else f"OK {msg}") -def warn(msg): print(f"{YEL}!{END} {msg}" if _tty() else f"! {msg}") -def fail(msg): print(f"{RED}✗{END} {msg}" if _tty() else f"FAIL {msg}") -def dim(msg): print(f"{DIM}{msg}{END}" if _tty() else msg) - -def prompt(question, default=None): - suffix = f" [{default}]" if default is not None else "" - while True: - try: - ans = input(f" {question}{suffix}: ").strip() - except EOFError: - print(); sys.exit(130) - if ans: - return ans - if default is not None: - return default - -def confirm(question, default=True): - suffix = ' [Y/n]' if default else ' [y/N]' - while True: - try: - ans = input(f" {question}{suffix}: ").strip().lower() - except EOFError: - print(); sys.exit(130) - if not ans: - return default - if ans in ('y', 'yes'): return True - if ans in ('n', 'no'): return False - - -# ---------- step 1: AC14K_M discovery ------------------------------------- -# Note: we deliberately do NOT do a bare TCP reachability probe before -# the real TLS handshake. The dryer's OCF stack treats a plain -# TCP-open-then-close (no TLS) as anomalous and enters a defensive state -# that closes subsequent handshakes' sockets immediately after CSM. -# Empirically observed; see commit history. Reachability is checked -# implicitly when we open TLS in step 3. - -def find_ac14km(): - """Look in ./certs/ for the AC14K_M cert + key under any of the - common filenames. Returns (cert_path, key_path) or (None, None).""" - cert_candidates = ['ac14k_m.pem', 'AC14K_M.pem', 'cert_1.pem'] - key_candidates = ['ac14k_m.key', 'AC14K_M.key', 'key.pem', 'ac14k_m_key.pem'] - cert = next((CERTS_DIR / n for n in cert_candidates if (CERTS_DIR / n).exists()), None) - key = next((CERTS_DIR / n for n in key_candidates if (CERTS_DIR / n).exists()), None) - return cert, key - - -def check_openssl(): - """Bootstrap shells out to openssl for cert generation — SHA-1 signing - was removed from python-cryptography in v43, and openssl is ubiquitous - enough that requiring it is reasonable.""" - if shutil.which('openssl') is None: - fail("openssl not found in PATH — required for cert generation") - return False - return True - - -def _run(cmd, **kw): - """Wrapper that surfaces stderr on failure.""" - res = subprocess.run(cmd, capture_output=True, text=True, **kw) - if res.returncode != 0: - raise RuntimeError( - f"`{' '.join(cmd)}` failed:\n{res.stderr.strip() or res.stdout.strip()}" - ) - return res - - -def _openssl_config(common_name, hub_uuid=None, include_samsung_role=True): - """Return an OpenSSL config snippet matching the proven canonical recipe - used to generate the original working `mega_chain.pem` for this project - (see spoof/mega_ext.cnf). All four SAN entries and the `clientAuth, - serverAuth` EKU values are defensive — the dryer's `memmem` scan only - cares about the Subject DN, but adjacent tooling reads the rest.""" - v3_lines = [ - "basicConstraints = CA:FALSE", - "keyUsage = digitalSignature, keyEncipherment", - f"extendedKeyUsage = clientAuth, serverAuth, {SAMSUNG_IOT_IDENTITY_OID}", - ] - if hub_uuid: - v3_lines.append("subjectAltName = @alt_names") - if include_samsung_role: - v3_lines.append( - f"{SAMSUNG_ROLE_OID} = ASN1:UTF8String:samsung.role.hub") - sections = [ - "[ req ]", - "distinguished_name = dn", - "prompt = no", - "req_extensions = v3", - "", - "[ dn ]", - f"CN = {common_name}", - "O = Samsung Electronics", - "C = KR", - "", - "[ v3 ]", - *v3_lines, - ] - if hub_uuid: - # Belt-and-braces SAN entries — three URI forms and a DNS name. - # Matches the canonical mega_ext.cnf exactly so the leaf is - # bit-for-bit equivalent to the cert known to authenticate. - sections += [ - "", - "[ alt_names ]", - f"URI.1 = urn:uuid:{hub_uuid}", - f"URI.2 = uri:uuid:{hub_uuid}", - f"URI.3 = uuid:{hub_uuid}", - f"DNS.1 = {hub_uuid}", - ] - return "\n".join(sections) + "\n" - - -def _generate_signed_cert(*, common_name, hub_uuid, include_samsung_role, - ca_cert, ca_key, out_key, out_cert, days): - """Generate an RSA-2048 key + SHA-1 signed cert via openssl.""" - with tempfile.TemporaryDirectory() as td: - tdp = Path(td) - conf = tdp / 'leaf.cnf' - csr = tdp / 'leaf.csr' - conf.write_text(_openssl_config(common_name, hub_uuid, - include_samsung_role)) - # 1) key + CSR with extensions baked into req_extensions - _run(['openssl', 'req', '-new', '-newkey', 'rsa:2048', '-nodes', - '-keyout', str(out_key), '-out', str(csr), '-config', str(conf)]) - # 2) sign with AC14K_M, SHA-1, copy the v3 extensions through - _run(['openssl', 'x509', '-req', '-in', str(csr), - '-CA', str(ca_cert), '-CAkey', str(ca_key), - '-CAcreateserial', '-out', str(out_cert), - '-days', str(days), '-sha1', - '-extfile', str(conf), '-extensions', 'v3']) - os.chmod(out_key, 0o600) - - -def generate_leaf(hub_uuid, ca_cert, ca_key, out_dir): - """The real leaf — Subject CN contains `urn:uuid:` so the - dryer's `memmem` scan recognises us as the SmartThings hub. Writes - mega.key and mega_chain.pem (leaf || AC14K_M).""" - subject_uri = f"urn:uuid:{hub_uuid}" - out_key = out_dir / 'mega.key' - out_leaf = out_dir / 'mega_leaf.pem' - out_chain = out_dir / 'mega_chain.pem' - _generate_signed_cert( - common_name=subject_uri, - hub_uuid=hub_uuid, - include_samsung_role=True, - ca_cert=ca_cert, ca_key=ca_key, - out_key=out_key, out_cert=out_leaf, - days=365 * 5, - ) - # Concatenate leaf || AC14K_M for the bridge's load_cert_chain. - out_chain.write_bytes(out_leaf.read_bytes() + Path(ca_cert).read_bytes()) - out_leaf.unlink() - return out_key, out_chain - - -def generate_probe(ca_cert, ca_key, tmp_dir): - """Throwaway leaf with NO `uuid:` in the Subject DN — the dryer treats - us as an anonymous-but-CA-trusted peer. Used once to attempt the - anonymous ACL read; never written to disk outside tmp_dir.""" - out_key = tmp_dir / 'probe.key' - out_leaf = tmp_dir / 'probe.pem' - out_chain = tmp_dir / 'probe_chain.pem' - _generate_signed_cert( - common_name='samsung-local-bootstrap-probe', - hub_uuid=None, - include_samsung_role=False, - ca_cert=ca_cert, ca_key=ca_key, - out_key=out_key, out_cert=out_leaf, - days=30, - ) - out_chain.write_bytes(out_leaf.read_bytes() + Path(ca_cert).read_bytes()) - out_leaf.unlink() - return out_key, out_chain - - -# ---------- step 4: anonymous ACL read ------------------------------------ - -def open_tls(host, port, cert_path, key_path, timeout=8): - """Same pattern as samsung_dryer.bridge._open_tls — drop OpenSSL 3.x - security level so SHA-1 leaves are accepted.""" - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.check_hostname = False - ctx.verify_mode = ssl.CERT_NONE - try: - ctx.set_ciphers('DEFAULT:@SECLEVEL=0') - except ssl.SSLError: - pass - ctx.load_cert_chain(certfile=str(cert_path), keyfile=str(key_path)) - raw = socket.create_connection((host, port), timeout=timeout) - sock = ctx.wrap_socket(raw) - sock.send(CSM) - sock.settimeout(2) - try: read_tcp(sock) - except (socket.timeout, ConnectionError): pass - sock.settimeout(timeout) - return sock - - -def coap_get(sock, path_segs, token=b'\x01\x02\x03\x04'): - opts = [(URI_PATH, s.encode()) for s in path_segs] - sock.send(enc_tcp(0x01, token=token, opts_b=enc_opts(opts))) - code, _tok, _opts, pl = read_tcp(sock) - return code, pl - - -def extract_hub_uuid_from_doxm(doxm_payload): - """Parse the CBOR-encoded /oic/sec/doxm response and return the hub - UUID. On this firmware, `devowneruuid` and `rowneruuid` both carry - the SmartThings hub's UUID — they're the same value in practice and - we prefer devowneruuid (the OCF spec field for the device's owner).""" - try: - doc = cbor2.loads(doxm_payload) - except Exception as e: - warn(f"doxm CBOR decode failed: {e}") - return None - if not isinstance(doc, dict): - warn(f"doxm decoded to {type(doc).__name__}, expected dict") - return None - for key in ('devowneruuid', 'rowneruuid'): - val = doc.get(key) - if isinstance(val, str) and looks_like_uuid(val): - return val - warn(f"doxm payload had no devowneruuid/rowneruuid (keys: " - f"{list(doc.keys())})") - return None - - -def try_anonymous_doxm_read(host, port, ca_cert, ca_key): - """Discover the hub UUID by reading /oic/sec/doxm anonymously. - - Mechanism: the dryer's baseline ACL contains a wildcard ACE - (`subjectuuid=*` perm=2) granting any authenticated peer read access - to /oic/sec/doxm. We don't need to be the hub — we just need to - complete a chain-valid TLS handshake. doxm.devowneruuid is the - SmartThings hub's UUID.""" - with tempfile.TemporaryDirectory() as td: - tdp = Path(td) - try: - key_path, chain_path = generate_probe(ca_cert, ca_key, tdp) - except RuntimeError as e: - warn(f"probe cert generation failed: {e}") - return None - try: - sock = open_tls(host, port, chain_path, key_path) - except ConnectionRefusedError: - fail(f"connection refused at {host}:{port} — wrong port, or " - f"the dryer isn't on the LAN.") - return None - except (ssl.SSLError, OSError) as e: - warn(f"anonymous TLS handshake failed: {e}") - return None - try: - code, pl = coap_get(sock, ['oic', 'sec', 'doxm']) - except ConnectionError as e: - warn(f"dryer closed the CoAP session immediately after CSM: {e}") - dim(" This usually means the dryer's OCF stack is in a " - "defensive cooldown — typically caused by a concurrent " - "TLS session (the bridge running) or rapid recent probes. " - "Stop main.py / the bridge container, wait ~60s, then re-run.") - return None - finally: - try: sock.close() - except Exception: pass - if code != 0x45: - warn(f"GET /oic/sec/doxm → {fmt_code(code)} (expected 2.05) " - f"— switching to manual entry") - return None - return extract_hub_uuid_from_doxm(pl) - - -# ---------- step 5: .env --------------------------------------------------- - -def maybe_write_env(appliance_ip, appliance_port): - env_path = REPO_ROOT / '.env' - example = REPO_ROOT / '.env.example' - if not example.exists(): - warn(".env.example missing — skipping .env generation") - return - if env_path.exists(): - if not confirm("Overwrite existing .env with new IP/port? (other " - "values preserved)", default=False): - dim(" leaving .env untouched") - return - text = example.read_text() - text = _replace_kv(text, 'APPLIANCE_IP', appliance_ip) - text = _replace_kv(text, 'APPLIANCE_OCF_PORT', str(appliance_port)) - env_path.write_text(text) - ok(f"wrote {env_path} — fill in MQTT_BROKER / MQTT_USER / MQTT_PASS before running main.py") - - -def _replace_kv(text, key, value): - out = [] - for line in text.splitlines(): - if line.startswith(f"{key}="): - out.append(f"{key}={value}") - else: - out.append(line) - return '\n'.join(out) + ('\n' if text.endswith('\n') else '') - - -# ---------- step 6: hub UUID validation ----------------------------------- - -def looks_like_uuid(s): - import re - return bool(re.fullmatch( - r'[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-' - r'[0-9a-fA-F]{4}-[0-9a-fA-F]{12}', s.strip())) - - -# ---------- main ---------------------------------------------------------- - -def main(): - print() - print(f"{BOLD}samsung-appliance-local — bootstrap{END}" if _tty() - else "samsung-appliance-local — bootstrap") - print(f"{DIM}This will discover your dryer, locate your CA cert, and " - f"generate the leaf used to authenticate as the SmartThings hub.{END}" - if _tty() else - "This will discover your dryer, locate your CA cert, and generate " - "the leaf used to authenticate as the SmartThings hub.") - print() - - # --- 1. dryer location --- - # Reachability is verified implicitly by the TLS handshake in step 3. - # We can't do a bare TCP probe here — that knocks the dryer's OCF - # session into a defensive state and breaks the subsequent TLS attempt. - info("Step 1 — dryer location") - appliance_ip = prompt("Dryer IP on your LAN", default=None) - appliance_port = int(prompt("OCF port (newer firmware uses 49154)", - default='49154')) - dim(f" Will connect to {appliance_ip}:{appliance_port} once we have " - f"a probe cert.") - print() - - # --- 2. AC14K_M --- - info("Step 2 — locate Samsung's AC14K_M intermediate CA") - CERTS_DIR.mkdir(parents=True, exist_ok=True) - cert_path, key_path = find_ac14km() - if cert_path is None or key_path is None: - fail(f"AC14K_M cert + key not found in {CERTS_DIR}/") - dim(f" Fetch them from: {AC14K_M_SOURCE}") - dim(f" Place as: {CERTS_DIR}/ac14k_m.pem and " - f"{CERTS_DIR}/ac14k_m.key (other common names accepted)") - return 2 - ok(f"found CA cert: {cert_path.name}") - ok(f"found CA key: {key_path.name}") - if not check_openssl(): - return 2 - print() - - # --- 3. hub UUID --- - info("Step 3 — discover your SmartThings hub UUID") - dim(" Reading /oic/sec/doxm anonymously — the dryer's baseline ACL") - dim(" allows any authenticated peer to read it (wildcard ACE).") - hub_uuid = try_anonymous_doxm_read(appliance_ip, appliance_port, - cert_path, key_path) - if hub_uuid: - ok(f"discovered hub UUID from /oic/sec/doxm: {hub_uuid}") - if not confirm("Use this UUID?", default=True): - hub_uuid = None - if not hub_uuid: - warn("Falling back to manual entry. Options B/C in the README " - "describe how to obtain it.") - while True: - hub_uuid = prompt("Hub UUID (8-4-4-4-12 hex)", default=None) - if looks_like_uuid(hub_uuid): - hub_uuid = hub_uuid.strip().lower() - break - warn("That doesn't look like a UUID. Format: " - "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx") - print() - - # --- 4. leaf --- - info("Step 4 — generate the leaf cert (mega.key + mega_chain.pem)") - mega_key = CERTS_DIR / 'mega.key' - mega_chain = CERTS_DIR / 'mega_chain.pem' - if mega_key.exists() or mega_chain.exists(): - warn(f"existing leaf cert detected in {CERTS_DIR}/") - if not confirm("Overwrite?", default=False): - dim(" leaving existing leaf in place — skipping generation") - print() - maybe_write_env(appliance_ip, appliance_port) - print() - ok("Done.") - return 0 - try: - key_out, chain_out = generate_leaf(hub_uuid, cert_path, key_path, - CERTS_DIR) - except RuntimeError as e: - fail(f"leaf cert generation failed: {e}") - return 2 - ok(f"wrote {key_out}") - ok(f"wrote {chain_out}") - print() - - # --- 5. .env --- - info("Step 5 — populate .env") - maybe_write_env(appliance_ip, appliance_port) - print() - - ok("Done. Next: edit .env to fill in MQTT_BROKER / MQTT_USER / " - "MQTT_PASS, then run main.py.") - return 0 - - -if __name__ == '__main__': - try: - sys.exit(main()) - except KeyboardInterrupt: - print(); sys.exit(130) diff --git a/deploy.sh b/deploy.sh index de36c29..1ae387c 100755 --- a/deploy.sh +++ b/deploy.sh @@ -5,7 +5,7 @@ # REMOTE_DIR — compose project (source code, .env, docker-compose.yml) # Convention: /mnt/user/compose/samsung-bridge/ # APPDATA_DIR — bind-mount source for /config inside the container -# (ab0b0ac4 client cert + key live here). +# (client cert + key live here). # Convention: /mnt/user/appdata/samsung-bridge/ # # The remote must already have the certs in $APPDATA_DIR. Run once @@ -13,7 +13,7 @@ # # source .env # ssh "$SSH_HOST" mkdir -p "$APPDATA_DIR" -# scp certs/ab0b0ac4_fullchain.pem certs/ab0b0ac4.key \ +# scp certs/client_fullchain.pem certs/client.key \ # "$SSH_HOST:$APPDATA_DIR/" # # Subsequent deploys (this script) ship source code + .env only; the @@ -63,13 +63,13 @@ ssh "${SSH_HOST}" "chmod 600 ${REMOTE_DIR}/.env" # Verify certs are present on the remote — they have to be uploaded # once before the first build. -if ! ssh "${SSH_HOST}" "test -s ${APPDATA_DIR}/ab0b0ac4_fullchain.pem && test -s ${APPDATA_DIR}/ab0b0ac4.key"; then +if ! ssh "${SSH_HOST}" "test -s ${APPDATA_DIR}/client_fullchain.pem && test -s ${APPDATA_DIR}/client.key"; then echo - echo "WARNING: ${APPDATA_DIR}/ab0b0ac4_fullchain.pem and ab0b0ac4.key not" + echo "WARNING: ${APPDATA_DIR}/client_fullchain.pem and client.key not" echo "found on the remote. The container will start but fail to" echo "connect to the appliance until you upload them, e.g.:" echo " ssh ${SSH_HOST} mkdir -p ${APPDATA_DIR}" - echo " scp certs/ab0b0ac4_fullchain.pem certs/ab0b0ac4.key ${SSH_HOST}:${APPDATA_DIR}/" + echo " scp certs/client_fullchain.pem certs/client.key ${SSH_HOST}:${APPDATA_DIR}/" echo fi diff --git a/docker-compose.yml b/docker-compose.yml index 6a0b127..769bb08 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,10 +8,9 @@ services: # broker on 1883). No ports to expose. volumes: - # Holds the ab0b0ac4 client cert + key. APPDATA_DIR comes from - # .env; on Unraid this is typically - # /mnt/user/appdata/smartthings-local/. Bare-metal dev falls - # back to ./certs alongside this compose file. + # Holds the client cert + key. APPDATA_DIR comes from .env; + # on Unraid this is typically /mnt/user/appdata/smartthings-local/. + # Bare-metal dev falls back to ./certs alongside this compose file. - ${APPDATA_DIR:-./certs}:/config:ro # All runtime config is in .env. env_file passes every variable diff --git a/requirements-bootstrap.txt b/requirements-bootstrap.txt index 952be50..2360d27 100644 --- a/requirements-bootstrap.txt +++ b/requirements-bootstrap.txt @@ -1,4 +1,4 @@ -# Setup-only deps. bootstrap.py reuses cbor2 to parse the dryer's ACL -# response and shells out to `openssl` for cert generation (so SHA-1 -# signing keeps working independent of python-cryptography's policy). +# Setup-only deps for setup_cert.py: shells out to `openssl` for SHA-1 +# signing (independent of python-cryptography's policy) and uses +# pyOpenSSL for the optional --test DTLS handshake. -r requirements.txt diff --git a/samsung_appliance/config.py b/samsung_appliance/config.py index 0edb89a..3da393a 100644 --- a/samsung_appliance/config.py +++ b/samsung_appliance/config.py @@ -62,8 +62,8 @@ class SharedConfig: @classmethod def from_env(cls) -> 'SharedConfig': return cls( - CERT_PATH=_resolve_cert('CERT_PATH', 'ab0b0ac4_fullchain.pem'), - KEY_PATH=_resolve_cert('KEY_PATH', 'ab0b0ac4.key'), + CERT_PATH=_resolve_cert('CERT_PATH', 'client_fullchain.pem'), + KEY_PATH=_resolve_cert('KEY_PATH', 'client.key'), MQTT_BROKER=os.getenv('MQTT_BROKER'), MQTT_PORT=int(os.getenv('MQTT_PORT', '1883')), MQTT_USER=os.getenv('MQTT_USER') or None, diff --git a/setup_cert.py b/setup_cert.py new file mode 100644 index 0000000..24f4abb --- /dev/null +++ b/setup_cert.py @@ -0,0 +1,492 @@ +#!/usr/bin/env python3 +""" +setup_cert.py — One-shot client cert generator for local DTLS-CoAP +access to Samsung Tizen/RT-OCF appliances on your LAN. + +Builds a client cert keyed to the identity that each appliance's factory +ACL already grants `perm=31` on `href=*`. Everything used at build time +is fetched live from public sources; nothing is hardcoded. + +Steps: + +1. Fetch the AC14K_M intermediate CA bundle (CA cert + key + upstream + chain) from a public mirror. +2. Open a TLS connection to a Samsung cloud endpoint, read its + server cert, and extract the `uuid:` token from the subject DN. +3. Generate a fresh RSA-2048 key pair (yours, not Samsung's). +4. Build a CSR with the UUID in CN, OU, and SAN. +5. Sign the CSR with AC14K_M using SHA-1, matching the on-device + trust hierarchy. +6. Assemble `.key`, `.pem`, `_fullchain.pem`. +7. With `--test`, DTLS-handshake to an appliance and GET + `/oic/sec/acl`; a 2.05 reply confirms the cert is accepted. + +Background: + +- The cloud-bridge UUID is published in Samsung's own TLS server cert + subject DN — anyone can read it with `openssl s_client`. +- TizenRT iotivity locates the peer UUID via `memmem(subject, "uuid:")`, + so the same UUID in any RDN works. +- The AC14K_M intermediate has been public for years and remains in + current firmware trust stores. + +Fallbacks if the live fetches fail: + + # Manual UUID lookup + openssl s_client -connect :443 -servername \\ + -showcerts < /dev/null 2>/dev/null \\ + | openssl x509 -noout -subject + UUID= python setup_cert.py ... + + # Manual AC14K_M bundle (point at any mirror) + AC14K_M_CERT_BUNDLE=/path/to/cert.pem python setup_cert.py + +Usage: + + python setup_cert.py + python setup_cert.py --test + TARGET_IP=192.168.1.1 python setup_cert.py --test + +Env overrides (all optional): + AC14K_M_CERT AC14K_M cert PEM (skip live fetch) + AC14K_M_KEY AC14K_M private key PEM + AC14K_M_CERT_BUNDLE combined PEM (key + 4 certs) + CHAIN_DIR dir containing cert_1..4.pem + BRAYSTORM_URL bundle source URL + UUID supply the UUID manually + OUT_DIR output dir (default ./certs/) + TARGET_IP device IP for --test + TARGET_PORT device port for --test (default 49154) +""" +import argparse +import os +import re +import socket +import ssl +import subprocess +import sys +import tempfile +import urllib.request +from pathlib import Path + + +SAMSUNG_HOST = 'connect-v2.samsungiotcloud.com' +SAMSUNG_PORT = 443 + +BRAYSTORM_URL = ( + 'https://raw.githubusercontent.com/brayStorm/samsung-appliance-token/main/cert.pem' +) + +BUNDLE_CERT_NAMES = ['ac14k_m.pem', 'cert_2.pem', 'cert_3.pem', 'cert_4.pem'] + + +def fetch_samsung_uuid(timeout=10): + """Return (uuid, server_cert_pem) or (None, None) on failure.""" + try: + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + with socket.create_connection((SAMSUNG_HOST, SAMSUNG_PORT), timeout=timeout) as raw: + with ctx.wrap_socket(raw, server_hostname=SAMSUNG_HOST) as s: + der = s.getpeercert(binary_form=True) + except Exception as e: + print(f"[!] Could not fetch Samsung cloud cert: {e}", file=sys.stderr) + return None, None + + tmp = tempfile.NamedTemporaryFile(suffix='.der', delete=False) + tmp.write(der); tmp.close() + try: + subj = subprocess.run( + ['openssl', 'x509', '-inform', 'DER', '-in', tmp.name, + '-noout', '-subject'], + capture_output=True, text=True, check=True).stdout + pem = subprocess.run( + ['openssl', 'x509', '-inform', 'DER', '-in', tmp.name], + capture_output=True, text=True, check=True).stdout + finally: + os.unlink(tmp.name) + + m = re.search(r'uuid:([0-9a-fA-F-]{36})', subj) + if not m: + print(f"[!] No `uuid:...` in subject: {subj.strip()}", file=sys.stderr) + return None, pem + return m.group(1).lower(), pem + + +def split_bundle_pem(text): + """Split a combined PEM into (key_pem, [cert_pem, ...]). + Expects 1 private key + 4 certificates (leaf + 3 upstream).""" + key_re = re.compile( + r'-----BEGIN (?:RSA )?PRIVATE KEY-----.*?-----END (?:RSA )?PRIVATE KEY-----', + re.DOTALL) + cert_re = re.compile( + r'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----', + re.DOTALL) + keys = key_re.findall(text) + certs = cert_re.findall(text) + if len(keys) != 1: + raise ValueError(f"expected 1 private key block, found {len(keys)}") + if len(certs) != 4: + raise ValueError(f"expected 4 certificate blocks, found {len(certs)}") + return keys[0] + '\n', [c + '\n' for c in certs] + + +def fetch_ac14k_bundle(dest_dir, timeout=15): + """Download and split the AC14K_M bundle. Returns + {ac14k_cert, ac14k_key, chain_dir} of paths in dest_dir.""" + url = os.environ.get('BRAYSTORM_URL', BRAYSTORM_URL) + print(f" Fetching AC14K_M bundle...") + try: + with urllib.request.urlopen(url, timeout=timeout) as resp: + data = resp.read().decode('utf-8', errors='replace') + except Exception as e: + raise RuntimeError(f"bundle fetch failed: {e}") from e + + key_pem, cert_pems = split_bundle_pem(data) + + dest = Path(dest_dir); dest.mkdir(parents=True, exist_ok=True) + key_path = dest / 'ac14k_m.key' + key_path.write_text(key_pem) + try: + os.chmod(key_path, 0o600) + except OSError: + pass + cert_paths = [] + for name, pem in zip(BUNDLE_CERT_NAMES, cert_pems): + p = dest / name + p.write_text(pem) + cert_paths.append(p) + (dest / 'cert_1.pem').write_text(cert_pems[0]) + + return { + 'ac14k_cert': cert_paths[0], + 'ac14k_key': key_path, + 'chain_dir': dest, + } + + +def verify_cert_key_pair(cert_path, key_path): + """Compare modulus to confirm cert and key pair.""" + def modulus(args): + out = subprocess.run( + ['openssl'] + args, capture_output=True, text=True, check=True).stdout + m = re.search(r'Modulus=([0-9A-Fa-f]+)', out) + return m.group(1) if m else None + try: + cm = modulus(['x509', '-noout', '-modulus', '-in', str(cert_path)]) + km = modulus(['rsa', '-noout', '-modulus', '-in', str(key_path)]) + except subprocess.CalledProcessError as e: + raise RuntimeError(f"openssl modulus extraction failed: {e.stderr}") from e + if not cm or not km: + raise RuntimeError("could not extract modulus from cert and/or key") + if cm != km: + raise RuntimeError( + f"AC14K_M cert and key do not pair (cert modulus != key modulus)") + + +def run(cmd, **kw): + return subprocess.run(cmd, check=True, capture_output=True, text=True, **kw) + + +def mint_cert(uuid, ac14k_cert, ac14k_key, chain_files, out_dir): + """Mint a fresh-keyed client cert with UUID in CN+OU+SAN, signed by + AC14K_M with SHA-1. Returns dict of output paths.""" + out = Path(out_dir); out.mkdir(parents=True, exist_ok=True) + paths = { + 'key': out / 'client.key', + 'csr': out / 'client.csr', + 'leaf': out / 'client.pem', + 'fullchain': out / 'client_fullchain.pem', + 'ext': out / 'ext.cnf', + 'srl': out / 'client.srl', + } + + paths['ext'].write_text(f"""basicConstraints = CA:FALSE +keyUsage = digitalSignature, keyEncipherment +extendedKeyUsage = clientAuth, serverAuth, 1.3.6.1.4.1.51414.0.1.2 +subjectAltName = @alt_names +1.3.6.1.4.1.51414.1.3 = ASN1:UTF8String:samsung.role.hub + +[alt_names] +URI.1 = urn:uuid:{uuid} +URI.2 = uri:uuid:{uuid} +URI.3 = uuid:{uuid} +DNS.1 = {uuid} +""") + + run(['openssl', 'genrsa', '-out', str(paths['key']), '2048']) + try: + os.chmod(paths['key'], 0o600) + except OSError: + pass + + subject = ( + f"/OU=uuid:{uuid}" + f"/CN=urn:uuid:{uuid}" + f"/O=Samsung Electronics" + f"/C=KR" + ) + run(['openssl', 'req', '-new', '-key', str(paths['key']), + '-out', str(paths['csr']), '-subj', subject]) + + run(['openssl', 'x509', '-req', '-in', str(paths['csr']), + '-CA', str(ac14k_cert), '-CAkey', str(ac14k_key), + '-CAcreateserial', '-CAserial', str(paths['srl']), + '-out', str(paths['leaf']), '-days', '3650', + '-extfile', str(paths['ext']), '-sha1']) + + parts = [paths['leaf'].read_text()] + for p in chain_files: + parts.append(Path(p).read_text()) + paths['fullchain'].write_text(''.join(parts)) + + return paths + + +def test_handshake(target_ip, target_port, cert_path, key_path): + """DTLS-handshake to a device and GET /oic/sec/acl. + 2.05 means the cert authenticated; 4.01 means it didn't.""" + try: + from OpenSSL import SSL + except ImportError: + print("[!] pyOpenSSL not installed — skipping connectivity test") + print(" Install with: pip install pyOpenSSL") + return None + + import time + + ctx = SSL.Context(SSL.DTLS_METHOD) + ctx.set_verify(SSL.VERIFY_NONE, lambda *a: True) + ctx.set_cipher_list(b'ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0') + ctx.use_certificate_chain_file(str(cert_path)) + ctx.use_privatekey_file(str(key_path)) + ctx.check_privatekey() + conn = SSL.Connection(ctx, None) + conn.set_connect_state(); conn.set_ciphertext_mtu(1200) + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(2) + dest = (target_ip, target_port) + + def split_dtls(buf): + o, out = 0, [] + while o + 13 <= len(buf): + L = int.from_bytes(buf[o+11:o+13], 'big'); end = o + 13 + L + if end > len(buf): break + out.append(buf[o:end]); o = end + return out + + print(f"[+] DTLS handshake to {target_ip}:{target_port}...") + t0 = time.time() + handshake_ok = False + while time.time() - t0 < 12: + try: + conn.do_handshake(); handshake_ok = True; break + except SSL.WantReadError: pass + except SSL.Error as e: + print(f" SSL error: {e}"); return False + try: + out = conn.bio_read(65535) + if out: + for r in split_dtls(out): sock.sendto(r, dest) + except SSL.WantReadError: pass + try: + data, _ = sock.recvfrom(65535) + if data: conn.bio_write(data) + except socket.timeout: pass + time.sleep(0.05) + + if not handshake_ok: + print(f" handshake TIMEOUT after {time.time()-t0:.1f}s") + sock.close(); return False + print(f" handshake OK in {time.time()-t0:.2f}s") + + msg = ( + bytes([0x41, 0x01, 0xab, 0x00, 0xaa]) + + bytes([0xb3]) + b'oic' + bytes([0x03]) + b'sec' + bytes([0x03]) + b'acl' + + bytes([0x61]) + b'\x3c' + ) + conn.send(msg) + try: + while True: + out = conn.bio_read(65535) + if not out: break + sock.sendto(out, dest) + except SSL.WantReadError: pass + + deadline = time.time() + 6 + while time.time() < deadline: + try: + data, _ = sock.recvfrom(65535) + if data: + conn.bio_write(data) + try: + pl = conn.recv(65535) + code = pl[1] + print(f" GET /oic/sec/acl -> {code>>5}.{code&0x1F:02d}") + if code == 0x45: + print(f" OK — cert accepted by the device ACL") + sock.close(); return True + else: + print(f" Unexpected response code") + sock.close(); return False + except SSL.WantReadError: continue + except socket.timeout: pass + time.sleep(0.05) + print(f" GET /oic/sec/acl TIMEOUT") + sock.close(); return False + + +def resolve_ac14k_inputs(out_dir): + """Return (ac14k_cert, ac14k_key, chain_files). + + Resolution order: env-supplied cert+key+chain dir, then env-supplied + combined bundle, then live fetch from BRAYSTORM_URL.""" + env_cert = os.environ.get('AC14K_M_CERT') + env_key = os.environ.get('AC14K_M_KEY') + env_dir = os.environ.get('CHAIN_DIR') + env_bundle = os.environ.get('AC14K_M_CERT_BUNDLE') + + if env_cert and env_key and env_dir: + print(f" Using AC14K_M materials from env vars") + for path, label in [(env_cert, 'AC14K_M_CERT'), (env_key, 'AC14K_M_KEY')]: + if not Path(path).is_file(): + raise FileNotFoundError(f"{label} not found: {path}") + chain = sorted(Path(env_dir).glob('cert_*.pem')) + if len(chain) < 4: + raise RuntimeError( + f"CHAIN_DIR needs cert_1..cert_4.pem (leaf + 3 upstream); " + f"found: {[p.name for p in chain]}") + return Path(env_cert), Path(env_key), chain + + bundle_dir = Path(out_dir) / '.bundle' + + if env_bundle: + print(f" Splitting AC14K_M bundle from {env_bundle}") + text = Path(env_bundle).read_text() + key_pem, cert_pems = split_bundle_pem(text) + bundle_dir.mkdir(parents=True, exist_ok=True) + key_path = bundle_dir / 'ac14k_m.key' + key_path.write_text(key_pem) + try: + os.chmod(key_path, 0o600) + except OSError: + pass + for name, pem in zip(BUNDLE_CERT_NAMES, cert_pems): + (bundle_dir / name).write_text(pem) + (bundle_dir / 'cert_1.pem').write_text(cert_pems[0]) + chain = sorted(bundle_dir.glob('cert_*.pem')) + return bundle_dir / 'ac14k_m.pem', key_path, chain + + try: + result = fetch_ac14k_bundle(bundle_dir) + except Exception as e: + msg = ( + f"\n[!] Could not fetch AC14K_M bundle: {e}\n" + f"\n Workarounds:\n" + f" - Point at a local PEM: AC14K_M_CERT_BUNDLE=/path/to/cert.pem python setup_cert.py\n" + f" - Point at a mirror: BRAYSTORM_URL=https:///cert.pem python setup_cert.py\n" + ) + print(msg, file=sys.stderr) + raise SystemExit(3) + chain = sorted(result['chain_dir'].glob('cert_*.pem')) + return result['ac14k_cert'], result['ac14k_key'], chain + + +def main(): + p = argparse.ArgumentParser( + formatter_class=argparse.RawDescriptionHelpFormatter, + description=__doc__) + p.add_argument('--test', action='store_true', + help='After minting, attempt a DTLS handshake to TARGET_IP:TARGET_PORT') + args = p.parse_args() + + out_dir = os.environ.get('OUT_DIR', './certs/') + target_ip = os.environ.get('TARGET_IP') + target_port = int(os.environ.get('TARGET_PORT', 49154)) + uuid_override = os.environ.get('UUID') + + print("=" * 60) + print("Phase 1: AC14K_M signing materials") + print("=" * 60) + try: + ac14k_cert, ac14k_key, chain_files = resolve_ac14k_inputs(out_dir) + except SystemExit: + raise + except Exception as e: + print(f"[!] {e}", file=sys.stderr) + return 2 + print(f" AC14K_M cert: {ac14k_cert}") + print(f" AC14K_M key: {ac14k_key}") + print(f" chain: {len(chain_files)} certs ({', '.join(p.name for p in chain_files)})") + + try: + verify_cert_key_pair(ac14k_cert, ac14k_key) + except RuntimeError as e: + print(f"[!] AC14K_M cert/key sanity check failed: {e}", file=sys.stderr) + return 2 + print(f" cert/key modulus pair OK") + + print() + print("=" * 60) + print("Phase 2: identify peer UUID") + print("=" * 60) + samsung_pem = None + if uuid_override: + uuid = uuid_override.lower() + print(f" Using UUID from env: {uuid}") + else: + print(f" Fetching from {SAMSUNG_HOST}:{SAMSUNG_PORT}...") + uuid, samsung_pem = fetch_samsung_uuid() + if uuid is None: + print(f"\n [!] Live fetch failed.", file=sys.stderr) + print(f"\n Workaround:", file=sys.stderr) + print(f" 1. From any machine with internet access, run:", file=sys.stderr) + print(f" openssl s_client -connect {SAMSUNG_HOST}:{SAMSUNG_PORT} \\", file=sys.stderr) + print(f" -servername {SAMSUNG_HOST} \\", file=sys.stderr) + print(f" -showcerts < /dev/null 2>/dev/null \\", file=sys.stderr) + print(f" | openssl x509 -noout -subject", file=sys.stderr) + print(f" 2. Find OU=uuid: in the subject.", file=sys.stderr) + print(f" 3. Re-run with UUID= ...", file=sys.stderr) + return 3 + print(f" Extracted UUID: {uuid}") + if samsung_pem: + samsung_ref = Path(out_dir); samsung_ref.mkdir(parents=True, exist_ok=True) + (samsung_ref / 'samsung_cloud_leaf.pem').write_text(samsung_pem) + print(f" Saved server leaf cert to " + f"{samsung_ref / 'samsung_cloud_leaf.pem'}") + + print() + print("=" * 60) + print(f"Phase 3: mint client cert with UUID {uuid}") + print("=" * 60) + paths = mint_cert(uuid, ac14k_cert, ac14k_key, chain_files, out_dir) + print(f" key: {paths['key']}") + print(f" leaf: {paths['leaf']}") + print(f" fullchain: {paths['fullchain']}") + + subj_out = run(['openssl', 'x509', '-in', str(paths['leaf']), '-noout', '-subject']) + print(f" Subject: {subj_out.stdout.strip().replace('subject=', '')}") + + if args.test: + print() + print("=" * 60) + print("Phase 4: verify cert against target appliance") + print("=" * 60) + if not target_ip: + print(" [!] TARGET_IP not set; cannot run connectivity test", file=sys.stderr) + else: + result = test_handshake(target_ip, target_port, paths['fullchain'], paths['key']) + if result is True: + print("\n Cert is functional. Drop fullchain.pem + key into your bridge config.") + elif result is False: + print("\n Cert failed verification. Check target IP/port and try again.") + + print() + print("=" * 60) + print("Done. Output dir:", Path(out_dir).resolve()) + print("=" * 60) + return 0 + + +if __name__ == '__main__': + sys.exit(main())