bhetherman/SmartThings-Local
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ship setup_cert.py to repo root, auto-fetch all CA materials

Browse Source 
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.
This commit is contained in:
Jack Nagy
2026-06-30 19:27:24 +01:00
parent 709fdf444d
commit 0374f8cf68
8 changed files with 532 additions and 527 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+3 -4
View File
@@ -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
README.md
+24 -29
View File
@@ -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 <samsung-host>:443 -servername <samsung-host> \
-showcerts < /dev/null 2>/dev/null \
| openssl x509 -noout -subject
# subject=C=KR, O=Samsung Electronics, OU=uuid:<UUID>, CN=*.samsungiotcloud.com
```
The UUID lives in `OU=uuid:<UUID>`. Samsung's cert is valid through **2035-04-09**.
The UUID lives in `OU=uuid:<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=<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=<uuid>` env, or supply the AC14K_M bundle via `AC14K_M_CERT_BUNDLE=/path/to/cert.pem`. `BRAYSTORM_URL=<mirror>` 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 `<prefix>/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
```
bootstrap.py
-480
View File
@@ -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:<HUB>) 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:<HUB_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)
deploy.sh
+5 -5
View File
@@ -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
docker-compose.yml
+3 -4
View File
@@ -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
requirements-bootstrap.txt
+3 -3
View File
@@ -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
samsung_appliance/config.py
+2 -2
View File
@@ -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,
setup_cert.py
+492
View File
@@ -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:<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 `<uuid>.key`, `<uuid>.pem`, `<uuid>_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 <samsung-host>:443 -servername <samsung-host> \\
-showcerts < /dev/null 2>/dev/null \\
| openssl x509 -noout -subject
UUID=<paste-uuid-here> 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://<mirror>/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:<UUID> in the subject.", file=sys.stderr)
print(f" 3. Re-run with UUID=<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())