bhetherman/SmartThings-Local
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b34f0e824889c49a6a700c1d4dc9b9461e8de430
SmartThings-Local/setup_cert.py
T
Jack Nagy 0374f8cf68 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.
2026-06-30 19:27:24 +01:00

493 lines
18 KiB
Python
Raw Blame History

#!/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())
Reference in New Issue View Git Blame Copy Permalink