From 44afe4f00a64047c3c7da7a6e4ad43007b9e19e0 Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Tue, 7 Jul 2026 22:19:22 -0500 Subject: [PATCH] Point README at upstream setup_cert.py instead of vendoring a copy Getting the AC14K_M CA cert+key is a protocol-layer concern, not an HA-integration concern, and the two copies here had already drifted from each other. Drop root setup_cert.py + requirements-bootstrap.txt and have Part 2 explain why the CA cert is needed, then link to the smartthings-local project's setup_cert.py as the canonical way to obtain it. --- README.md | 13 +- requirements-bootstrap.txt | 4 - setup_cert.py | 492 ------------------------------------- 3 files changed, 4 insertions(+), 505 deletions(-) delete mode 100644 requirements-bootstrap.txt delete mode 100644 setup_cert.py diff --git a/README.md b/README.md index f2907c9..65d2802 100644 --- a/README.md +++ b/README.md @@ -47,16 +47,13 @@ nmap -Pn -sU -p 49152-49160 "$APPLIANCE_IP" ## Part 2 — One-time: get the AC14K_M CA credentials -The integration authenticates with a client cert chained to `AC14K_M`, an intermediate CA that's been public for years and remains in current firmware trust stores. Every Samsung Tizen/RT-OCF appliance's factory ACL grants the identity in that chain full CRUDN access, so the same CA can mint a working cert for any appliance on your LAN — HA does the per-device minting itself once you give it the CA. +The config flow (Part 3) needs a **CA certificate and CA private key** to mint each device's leaf cert itself. Specifically, it needs the `AC14K_M` intermediate CA: a cert chain that's been public for years and still ships in current Samsung firmware trust stores. It's required because every Samsung Tizen/RT-OCF appliance's factory ACL grants full CRUDN access (`perm=31` on `href=*`) to whatever identity is chained to that CA — so a cert signed by it is the one thing that lets HA talk to your appliance without Samsung's cloud in the loop. HA doesn't need the *device's* original cert or key, just something `AC14K_M` has signed, and it mints that itself once you give it the CA. -```sh -pip install -r requirements-bootstrap.txt -python setup_cert.py -``` +This repo doesn't vendor a script to fetch that CA bundle — the canonical way to obtain it lives in the `smartthings-local` protocol project: -This fetches the AC14K_M CA cert + key + upstream chain from a public mirror and writes them to `./certs/ac14k_m.pem` and `./certs/ac14k_m.key`. Nothing device-specific happens at this step — no IP, no handshake needed. (Pass `--test` with `TARGET_IP=` set if you want to sanity-check connectivity against a real device before touching HA at all.) +**[`setup_cert.py`](https://github.com/QuiteYellow/SmartThings-Local/blob/main/setup_cert.py)** -If the live fetch fails, the script prints an inline workaround: supply `AC14K_M_CERT_BUNDLE=/path/to/cert.pem`, or point `BRAYSTORM_URL=` at an alternate source. +Run that script (see its own usage docs) to fetch and verify the AC14K_M cert + key. It writes `ac14k_m.pem` (cert) and `ac14k_m.key` (key) — paste the contents of both into the HA config flow's "CA Certificate (PEM)" and "CA Private Key (PEM)" fields in Part 3. You only need to do this once; every appliance you add afterward reuses the same stored CA. ### Why this works @@ -129,8 +126,6 @@ custom_components/localthings/ by_type/ One DeviceRegistry per appliance type, composed from capabilities/ tests/ 80+ tests: registry composition, discovery, entity descriptors, golden-file regression against captured device dumps -setup_cert.py One-shot AC14K_M CA bundle fetcher (Part 2) -requirements-bootstrap.txt Deps for setup_cert.py only requirements-dev.txt Test deps, including the smartthings-local package docker-compose.yml / ha_config/ Local HA dev environment ``` diff --git a/requirements-bootstrap.txt b/requirements-bootstrap.txt deleted file mode 100644 index 2360d27..0000000 --- a/requirements-bootstrap.txt +++ /dev/null @@ -1,4 +0,0 @@ -# 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/setup_cert.py b/setup_cert.py deleted file mode 100644 index 24f4abb..0000000 --- a/setup_cert.py +++ /dev/null @@ -1,492 +0,0 @@ -#!/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())