#!/usr/bin/env python3 """ Planet Security Defense SDK (Python reference implementation) Patent : PCT WO 2025/127469 A1 (inventor/applicant: 이정훈, 2025-06-19) Site : https://planet.winnerbrothers.org Contact: official@winnerbrothers.org Use cases --------- - Drone / UAV / UGV / USV C2 over MAVLink, ROS2, custom radio - Satellite / ground-station authentication - V2X (K2 tank, K9 SPH, soldier-worn radio) - C4ISR command channel (KJCCS, JTACS, etc.) This SDK speaks the Defense API at /api/v1/defense/* — a thin wrapper around HTTP that takes care of: - 3-Round mutual authentication (Theorem 2) - HMAC-signed commands with replay/skew protection (Theorems 1, 6) - ECDH P-256 ephemeral key exchange (Theorem 4 forward secrecy) - CSPRNG via `secrets` (FIPS 140-3 path) Usage ----- from planet_defense_sdk import PlanetDefense pd = PlanetDefense( base_url="https://planet.winnerbrothers.org", secret_key="sk_xxx...", ) # 1) Provision GCS and UAV (one-time, secure channel) gcs = pd.keygen(callsign="GCS-01", classification="gcs", environment="lan") uav = pd.keygen(callsign="EAGLE-01", classification="uav", environment="lan") # 2) Establish 3-Round mutual auth session = pd.handshake(initiator=gcs, responder=uav) # 3) GCS sends signed command signed = pd.sign_command(session, sender=gcs, command="TAKEOFF") result = pd.verify_command(signed) assert result["valid"], result # 4) Send another (replay protection automatic) signed = pd.sign_command(session, sender=gcs, command="HOVER 50m") pd.verify_command(signed) """ import json import secrets import hashlib import hmac import time import urllib.request import urllib.error from typing import Any, Optional try: # cryptography is the standard library for ECDH; install via: # pip install cryptography from cryptography.hazmat.primitives.asymmetric.ec import ( generate_private_key, SECP256R1, ECDH, ) from cryptography.hazmat.primitives.serialization import ( Encoding, PublicFormat, ) HAS_CRYPTO = True except ImportError: HAS_CRYPTO = False # ───────────────────────────────────────────────────────────────────────────── # HTTP client (no third-party dependencies for the core path) # ───────────────────────────────────────────────────────────────────────────── class PlanetDefense: def __init__(self, base_url: str, secret_key: str, timeout: float = 10.0): if not secret_key.startswith("sk_"): raise ValueError("Defense API requires a secret key (sk_...)") self.base_url = base_url.rstrip("/") self.secret_key = secret_key self.timeout = timeout # ── HTTP helper ───────────────────────────────────────────────────────── def _post(self, path: str, body: dict) -> dict: return self._req("POST", path, body) def _get(self, path: str) -> dict: return self._req("GET", path, None) def _req(self, method: str, path: str, body: Optional[dict]) -> dict: url = f"{self.base_url}{path}" data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request( url, data=data, method=method, headers={ "Authorization": f"Bearer {self.secret_key}", "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "PlanetDefenseSDK/1.0 (python)", }, ) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: try: return {"error_status": e.code, **json.loads(e.read().decode("utf-8"))} except Exception: return {"error_status": e.code, "error": str(e)} # ── Status ────────────────────────────────────────────────────────────── def status(self) -> dict: return self._get("/api/v1/defense/status") # ── Keygen ────────────────────────────────────────────────────────────── def keygen( self, *, callsign: str, classification: str, environment: str, owner_org: Optional[str] = None, hsm_attested: bool = False, ) -> dict: body = { "callsign": callsign, "classification": classification, "environment": environment, "hsmAttested": hsm_attested, } if owner_org is not None: body["ownerOrg"] = owner_org return self._post("/api/v1/defense/keygen", body) # ── 3-Round handshake (compound) ──────────────────────────────────────── def handshake(self, *, initiator: dict, responder: dict) -> dict: """ Run all 3 rounds of mutual authentication. Returns: {handshakeId, sessionKey, deltaMs, ...} """ if not HAS_CRYPTO: raise RuntimeError( "ECDH ephemeral keys require `cryptography` package. " "Install: pip install cryptography" ) # Generate ECDH P-256 ephemeral keys for both sides init_priv = generate_private_key(SECP256R1()) resp_priv = generate_private_key(SECP256R1()) init_pub_hex = init_priv.public_key().public_bytes( encoding=Encoding.X962, format=PublicFormat.UncompressedPoint ).hex() resp_pub_hex = resp_priv.public_key().public_bytes( encoding=Encoding.X962, format=PublicFormat.UncompressedPoint ).hex() # Round 1 — initiator commit init_pub_commit = hashlib.sha256(init_pub_hex.encode("utf-8")).hexdigest() r1 = self._post("/api/v1/defense/handshake/init", { "initiatorDeviceId": initiator["device"]["deviceId"], "responderDeviceId": responder["device"]["deviceId"], "initEphemeralPubHash": init_pub_commit, }) if not r1.get("success"): raise RuntimeError(f"Round 1 failed: {r1}") hs_id = r1["handshakeId"] # Round 2 — responder reply r2 = self._post("/api/v1/defense/handshake/respond", { "handshakeId": hs_id, "respondEphemeralPub": resp_pub_hex, }) if not r2.get("success"): raise RuntimeError(f"Round 2 failed: {r2}") # Round 3 — finalize r3 = self._post("/api/v1/defense/handshake/finalize", { "handshakeId": hs_id, "initEphemeralPub": init_pub_hex, }) if not r3.get("success"): raise RuntimeError(f"Round 3 failed: {r3}") return { "handshakeId": hs_id, "sessionKey": r3["sessionKey"], "deltaMs": r3.get("deltaMs"), "initiator": initiator, "responder": responder, } # ── Command sign / verify ─────────────────────────────────────────────── def sign_command(self, session: dict, *, sender: dict, command: str) -> dict: """Server-side signs and returns the bundle to transmit.""" r = self._post("/api/v1/defense/command/sign", { "handshakeId": session["handshakeId"], "senderDeviceId": sender["device"]["deviceId"], "command": command, }) if not r.get("success"): raise RuntimeError(f"sign_command failed: {r}") return r["signedCommand"] def verify_command(self, signed_command: dict) -> dict: return self._post("/api/v1/defense/command/verify", signed_command) # ── Heartbeat (passive or verified) ───────────────────────────────────── def heartbeat( self, *, device_id: str, state_hash: Optional[str] = None, nonce: Optional[str] = None, timestamp: Optional[int] = None, ) -> dict: body: dict = {"deviceId": device_id} if state_hash and nonce and timestamp is not None: body.update(stateHash=state_hash, nonce=nonce, timestamp=timestamp) return self._post("/api/v1/defense/heartbeat", body) # ── Local HMAC verification helpers (for offline-deployable receivers) ── @staticmethod def local_hmac_verify( session_key_hex: str, signed_command: dict ) -> bool: """ After the receiver has the sessionKey (via secure provisioning), it can verify HMAC offline without round-tripping to the server. """ msg = ( f"{signed_command['command']}|{signed_command['senderStateHash']}|" f"{signed_command['nonce']}|{signed_command['timestamp']}|" f"{signed_command['commandId']}" ).encode("utf-8") key = bytes.fromhex(session_key_hex) expected = hmac.new(key, msg, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signed_command["signature"]) @staticmethod def now_ms() -> int: return int(time.time() * 1000) @staticmethod def gen_nonce(byte_len: int = 16) -> str: return secrets.token_hex(byte_len) # ───────────────────────────────────────────────────────────────────────────── # Self-test (run: python planet_defense_sdk.py) # ───────────────────────────────────────────────────────────────────────────── def _self_test(base_url: str, secret_key: str) -> None: print(f"[*] Connecting: {base_url}") pd = PlanetDefense(base_url, secret_key) print("[*] Status:") s = pd.status() print(f" serverTime={s.get('serverTime')} bits={s.get('capability', {}).get('classicalSecurityBits')}") print("[*] Provisioning GCS + UAV...") gcs = pd.keygen(callsign="GCS-TEST", classification="gcs", environment="lan") uav = pd.keygen(callsign="UAV-TEST", classification="uav", environment="lan") print(f" GCS deviceId={gcs['device']['deviceId']}") print(f" UAV deviceId={uav['device']['deviceId']}") print("[*] 3-Round mutual auth...") session = pd.handshake(initiator=gcs, responder=uav) print(f" sessionKey[:16]={session['sessionKey'][:16]} deltaMs={session['deltaMs']}") print("[*] Signing TAKEOFF...") signed = pd.sign_command(session, sender=gcs, command="TAKEOFF altitude=50m") result = pd.verify_command(signed) assert result.get("valid"), f"verify failed: {result}" print(f" [OK] commandId={result['commandId']}") print("[*] Replay attack (resend same signedCommand)...") replay = pd.verify_command(signed) assert replay.get("valid") is False, f"replay should fail: {replay}" print(f" [OK] rejected: {replay.get('reason')}") print("[*] Local HMAC verify...") ok = PlanetDefense.local_hmac_verify(session["sessionKey"], signed) print(f" local HMAC: {'OK' if ok else 'FAIL'}") print("[*] Heartbeat (passive)...") hb = pd.heartbeat(device_id=uav["device"]["deviceId"]) print(f" healthy={hb.get('healthy')} skew_ms={hb.get('skewMs', 'n/a')}") print("\n[OK] All self-tests passed.") if __name__ == "__main__": import argparse p = argparse.ArgumentParser() p.add_argument("--base-url", default="http://localhost:3000") p.add_argument("--secret-key", required=True, help="sk_...") args = p.parse_args() _self_test(args.base_url, args.secret_key)