← Public packages
@kentcdodds/x
X API v2 helpers for tweets, search, legacy DMs, and encrypted X Chat via a Fly XDK sidecar.
src/sidecar-source.ts
63 lines · 13.3 KB · TypeScriptimport {
DEFAULT_SIDECAR_URL,
FLY_APP_NAME,
SIDECAR_URL_VALUE,
buildFlyToml,
sidecarUrlFromApp,
} from './owner-config.ts'
import { secretSetupUrls } from './sidecar.ts'
const files = {
"Dockerfile": "FROM python:3.12-slim\nWORKDIR /app\nCOPY requirements.txt .\nRUN pip install --no-cache-dir -r requirements.txt\nCOPY server.py .\nENV PORT=8080\nEXPOSE 8080\nCMD [\"python\", \"server.py\"]\n",
"fly.toml": buildFlyToml(),
"requirements.txt": "chatxdk==0.5.0\n",
"server.py": "\"\"\"On-demand X Chat XDK sidecar.\n\nUnlocks the existing Juicebox identity with the user's PIN (never generates a\nnew keypair) and decrypts/encrypts events for the owning Kody X package.\n\nAuth: `Authorization: Bearer $SIDECAR_TOKEN`\nPIN: `X-Chat-Pin` header (never logged, never sent to api.x.com)\n\"\"\"\n\nfrom __future__ import annotations\n\nimport base64\nimport hashlib\nimport hmac\nimport json\nimport os\nimport threading\nfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServer\nfrom typing import Any\n\nfrom chat_xdk import Chat\n\ntry:\n from chat_xdk import guesses_remaining as juicebox_guesses_remaining\nexcept ImportError:\n def juicebox_guesses_remaining(exc: BaseException) -> int | None:\n text = str(exc)\n marker = 'guesses_remaining='\n index = text.find(marker)\n if index < 0:\n return None\n raw = text[index + len(marker) :].split()[0].rstrip(',;')\n try:\n return int(raw)\n except ValueError:\n return None\n\nPORT = int(os.environ.get(\"PORT\", \"8080\"))\nSIDECAR_TOKEN = os.environ.get(\"SIDECAR_TOKEN\", \"\")\n\n_sessions: dict[str, Chat] = {}\n_session_lock = threading.Lock()\n\n\ndef _jsonable(value: Any) -> Any:\n if isinstance(value, dict):\n return {str(key): _jsonable(item) for key, item in value.items()}\n if isinstance(value, (list, tuple)):\n return [_jsonable(item) for item in value]\n if isinstance(value, (bytes, bytearray)):\n return base64.b64encode(bytes(value)).decode(\"ascii\")\n if hasattr(value, \"model_dump\"):\n return _jsonable(value.model_dump())\n if isinstance(value, (str, int, float, bool)) or value is None:\n return value\n return str(value)\n\n\ndef _digest(*parts: str) -> str:\n hasher = hashlib.sha256()\n for part in parts:\n hasher.update(part.encode(\"utf-8\"))\n hasher.update(b\"\\0\")\n return hasher.hexdigest()\n\n\ndef _config_json(juicebox_config: Any) -> str:\n if isinstance(juicebox_config, str) and juicebox_config.strip():\n return juicebox_config\n if isinstance(juicebox_config, dict):\n return json.dumps(juicebox_config)\n raise ValueError(\"juicebox_config is required\")\n\n\ndef _as_string_list(value: Any) -> list[str]:\n if not isinstance(value, list):\n return []\n out: list[str] = []\n for item in value:\n if isinstance(item, str) and item:\n out.append(item)\n elif isinstance(item, dict):\n encoded = item.get(\"encoded_event\") or item.get(\"encodedEvent\")\n if isinstance(encoded, str) and encoded:\n out.append(encoded)\n return out\n\n\ndef _unlock(user_id: str, pin: str, juicebox_config: Any, signing_key_version: str) -> Chat:\n config_json = _config_json(juicebox_config)\n cache_key = _digest(user_id, signing_key_version, pin, config_json)\n with _session_lock:\n existing = _sessions.get(cache_key)\n if existing is not None and existing.is_unlocked():\n return existing\n chat = Chat(config_json)\n chat.unlock(pin)\n chat.set_identity(user_id, signing_key_version)\n chat.set_cache_keys(True)\n _sessions.clear()\n _sessions[cache_key] = chat\n return chat\n\n\ndef _prepare_chat(body: dict[str, Any], pin: str) -> Chat:\n user_id = str(body.get(\"user_id\") or \"\").strip()\n if not user_id:\n raise ValueError(\"user_id is required\")\n signing_key_version = str(body.get(\"signing_key_version\") or \"\").strip()\n if not signing_key_version:\n raise ValueError(\"signing_key_version is required\")\n chat = _unlock(user_id, pin, body.get(\"juicebox_config\"), signing_key_version)\n signing_keys = body.get(\"signing_keys\")\n if isinstance(signing_keys, list) and signing_keys:\n chat.set_signing_keys(signing_keys)\n if hasattr(chat, \"set_reject_unverified\"):\n chat.set_reject_unverified(body.get(\"reject_unverified\") is not False)\n return chat\n\n\ndef _send_payload(payload: Any) -> dict[str, Any]:\n return {\n \"message_id\": getattr(payload, \"message_id\", None),\n \"encoded_message_create_event\": getattr(payload, \"encrypted_content\", None),\n \"encoded_message_event_signature\": getattr(payload, \"encoded_event_signature\", None),\n \"conversation_key_version\": getattr(payload, \"conversation_key_version\", None),\n \"should_notify\": getattr(payload, \"should_notify\", None),\n }\n\n\ndef _authorized(handler: BaseHTTPRequestHandler) -> bool:\n if not SIDECAR_TOKEN:\n return False\n header = handler.headers.get(\"Authorization\", \"\")\n scheme, _, token = header.partition(\" \")\n if scheme.lower() != \"bearer\" or not token:\n return False\n return hmac.compare_digest(token, SIDECAR_TOKEN)\n\n\nclass Handler(BaseHTTPRequestHandler):\n server_version = \"kody-x-chat/1\"\n\n def log_message(self, fmt: str, *args: Any) -> None:\n path = self.path.split(\"?\", 1)[0]\n super().log_message(\"%s %s\", path, fmt % args)\n\n def _write(self, status: int, payload: dict[str, Any]) -> None:\n body = json.dumps(payload).encode(\"utf-8\")\n self.send_response(status)\n self.send_header(\"Content-Type\", \"application/json\")\n self.send_header(\"Content-Length\", str(len(body)))\n self.end_headers()\n self.wfile.write(body)\n\n def _read_json(self) -> dict[str, Any]:\n length = int(self.headers.get(\"Content-Length\", \"0\") or \"0\")\n raw = self.rfile.read(length) if length else b\"{}\"\n parsed = json.loads(raw.decode(\"utf-8\") or \"{}\")\n if not isinstance(parsed, dict):\n raise ValueError(\"JSON object body required\")\n return parsed\n\n def do_GET(self) -> None:\n path = self.path.split(\"?\", 1)[0]\n if path == \"/health\":\n self._write(\n 200,\n {\n \"ok\": True,\n \"service\": \"kody-x-chat\",\n \"unlocked\": any(chat.is_unlocked() for chat in _sessions.values()),\n },\n )\n return\n self._write(404, {\"error\": \"not_found\"})\n\n def do_POST(self) -> None:\n path = self.path.split(\"?\", 1)[0]\n if not _authorized(self):\n self._write(401, {\"error\": \"unauthorized\"})\n return\n pin = (self.headers.get(\"X-Chat-Pin\") or \"\").strip()\n if not pin:\n self._write(400, {\"error\": \"missing_pin\"})\n return\n try:\n body = self._read_json()\n except Exception:\n self._write(400, {\"error\": \"invalid_json\"})\n return\n try:\n if path == \"/v1/decrypt-events\":\n self._write(200, self._decrypt(body, pin))\n return\n if path == \"/v1/encrypt-message\":\n self._write(200, self._encrypt(body, pin))\n return\n except ValueError as error:\n self._write(400, {\"error\": \"invalid_request\", \"message\": str(error)})\n return\n except Exception as error:\n remaining = juicebox_guesses_remaining(error)\n payload: dict[str, Any] = {\"error\": \"sidecar_failed\", \"message\": str(error)}\n if remaining is not None:\n payload[\"guesses_remaining\"] = remaining\n self._write(500, payload)\n return\n self._write(404, {\"error\": \"not_found\"})\n\n def _decrypt(self, body: dict[str, Any], pin: str) -> dict[str, Any]:\n chat = _prepare_chat(body, pin)\n encoded_events = _as_string_list(body.get(\"encoded_events\"))\n if not encoded_events:\n raise ValueError(\"encoded_events is required\")\n result = chat.decrypt_events(encoded_events)\n messages = []\n for item in result.get(\"messages\") or []:\n event = item.get(\"event\") if isinstance(item, dict) else None\n messages.append({\"event\": _jsonable(event or item)})\n errors = result.get(\"errors\") or {}\n if errors and body.get(\"reject_unverified\") is False:\n conversation_keys = {}\n try:\n extracted = chat.extract_conversation_keys(encoded_events)\n conversation_keys = extracted.get(\"keys\") or {}\n except Exception:\n conversation_keys = {}\n if conversation_keys:\n retried_errors: dict[str, str] = {}\n recovered: dict[int, dict[str, Any]] = {}\n for index_str, message in errors.items():\n try:\n index = int(index_str)\n event = chat.decrypt_event(\n encoded_events[index], conversation_keys=conversation_keys\n )\n recovered[index] = {\"event\": _jsonable(event)}\n except Exception as retry_error:\n retried_errors[index_str] = str(retry_error)\n for index in sorted(recovered):\n messages.append(recovered[index])\n errors = retried_errors\n return {\n \"ok\": True,\n \"messages\": messages,\n \"error_count\": len(errors) if isinstance(errors, dict) else 0,\n \"errors\": _jsonable(errors) if errors else {},\n }\n\n def _encrypt(self, body: dict[str, Any], pin: str) -> dict[str, Any]:\n chat = _prepare_chat(body, pin)\n warmup = _as_string_list(body.get(\"encoded_events\"))\n if warmup:\n chat.decrypt_events(warmup)\n conversation_id = str(body.get(\"conversation_id\") or \"\").strip()\n text = str(body.get(\"text\") or \"\")\n if not conversation_id:\n raise ValueError(\"conversation_id is required\")\n if not text:\n raise ValueError(\"text is required\")\n try:\n payload = chat.encrypt_message(conversation_id, text)\n except Exception:\n keys = {}\n if warmup:\n try:\n keys = chat.extract_conversation_keys(warmup).get(\"keys\") or {}\n except Exception:\n keys = {}\n if not keys:\n raise\n version = max(keys, key=lambda item: int(item) if item.isdigit() else -1)\n payload = chat.encrypt_message(\n conversation_id,\n text,\n conversation_key=keys[version],\n conversation_key_version=version,\n )\n return {\"ok\": True, \"payload\": _send_payload(payload)}\n\n\ndef main() -> None:\n if not SIDECAR_TOKEN:\n raise SystemExit(\"SIDECAR_TOKEN is required\")\n server = ThreadingHTTPServer((\"0.0.0.0\", PORT), Handler)\n server.serve_forever()\n\n\nif __name__ == \"__main__\":\n main()\n",
} as const
/**
* Fly/Docker source for the X Chat XDK sidecar.
* Other packages should import list/get/send chat helpers, not this sidecar.
* After a fork, pass `app` so fly.toml targets an app you own, then store
* `xChatSidecarUrl` in this package (see `./adapt`).
*
* @example
* import sidecarSource from 'kody:@kentcdodds/x/sidecar/source'
* const source = await sidecarSource({ app: 'your-x-chat' })
* // => { name: 'your-x-chat', url: 'https://your-x-chat.fly.dev', files: { ... } }
*/
export default async function sidecarSource(params: { app?: string } = {}) {
const app = params.app?.trim() || FLY_APP_NAME
const url = app === FLY_APP_NAME ? DEFAULT_SIDECAR_URL : sidecarUrlFromApp(app)
const host = new URL(url).host
return {
name: app,
description:
'Fly HTTP sidecar running the official Python XDK. Unlocks the existing X Chat identity with the user PIN and decrypts/encrypts events for the owning X package.',
host,
url,
env: {
SIDECAR_TOKEN: 'Required shared bearer. Save as user secret xChatSidecarToken with this Fly host approved.',
PORT: '8080',
},
files: {
...files,
'fly.toml': buildFlyToml(app),
},
secretSetup: secretSetupUrls(url),
next: {
persistStorage: {
export: './migrate-from-values',
params: { sidecarUrl: url },
},
adaptExport: './adapt',
},
deployNotes: [
'After a fork, pass `app` so this is not the listing owner’s Fly app.',
'Do not generate a new X Chat keypair. Unlock the existing identity with xChatPin.',
'X OAuth tokens stay on api.x.com. The sidecar receives only the PIN, juicebox_config, signing keys, and ciphertext.',
'Set SIDECAR_TOKEN on the Fly app and save the same value as xChatSidecarToken.',
`Store ${SIDECAR_URL_VALUE}=${url} via ./migrate-from-values so chat helpers stop using the listing default.`,
'Approve only the Fly sidecar host on xChatPin and xChatSidecarToken.',
],
}
}