Add 1Password Connect support for secrets
All checks were successful
Build and Push Container / build (push) Successful in 31s

Secrets can now use op:// references which are resolved via
1Password Connect API at startup. Set OP_CONNECT_HOST and
OP_CONNECT_TOKEN to enable.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Debian
2026-01-04 22:40:47 +00:00
parent 8ea35663d4
commit d728a8e235

View File

@@ -1,9 +1,69 @@
import os
import base64
import re
import requests
from dataclasses import dataclass
from typing import Optional
def fetch_from_op_connect(op_ref: str, connect_host: str, connect_token: str) -> str:
"""Fetch a secret from 1Password Connect using op:// reference."""
match = re.match(r"op://([^/]+)/([^/]+)/(.+)", op_ref)
if not match:
raise ValueError(f"Invalid 1Password reference: {op_ref}")
vault_name, item_name, field_name = match.groups()
headers = {"Authorization": f"Bearer {connect_token}"}
# Get vault ID
vaults_resp = requests.get(f"{connect_host}/v1/vaults", headers=headers)
vaults_resp.raise_for_status()
vault_id = next((v["id"] for v in vaults_resp.json() if v["name"] == vault_name), None)
if not vault_id:
raise ValueError(f"Vault not found: {vault_name}")
# Get item
items_resp = requests.get(
f"{connect_host}/v1/vaults/{vault_id}/items",
headers=headers,
params={"filter": f'title eq "{item_name}"'}
)
items_resp.raise_for_status()
items = items_resp.json()
if not items:
raise ValueError(f"Item not found: {item_name}")
# Get item details with fields
item_resp = requests.get(
f"{connect_host}/v1/vaults/{vault_id}/items/{items[0]['id']}",
headers=headers
)
item_resp.raise_for_status()
item = item_resp.json()
# Find field
for field in item.get("fields", []):
if field.get("label", "").lower() == field_name.lower() or field.get("id", "").lower() == field_name.lower():
return field.get("value", "")
raise ValueError(f"Field not found: {field_name}")
def resolve_secret(env_var: str, default: str = "") -> str:
"""Resolve a secret from env var or 1Password Connect."""
value = os.environ.get(env_var, default)
if value.startswith("op://"):
connect_host = os.environ.get("OP_CONNECT_HOST")
connect_token = os.environ.get("OP_CONNECT_TOKEN")
if not connect_host or not connect_token:
raise ValueError(f"{env_var} uses op:// reference but OP_CONNECT_HOST/OP_CONNECT_TOKEN not set")
return fetch_from_op_connect(value, connect_host, connect_token)
return value
@dataclass
class Config:
"""Application configuration loaded from environment variables."""
@@ -17,7 +77,7 @@ class Config:
@classmethod
def from_env(cls) -> "Config":
"""Load configuration from environment variables."""
ssh_key_b64 = os.environ.get("SSH_PRIVATE_KEY", "")
ssh_key_b64 = resolve_secret("SSH_PRIVATE_KEY", "")
# Decode base64 SSH key
try:
@@ -28,8 +88,8 @@ class Config:
return cls(
ssh_private_key=ssh_private_key,
uptime_kuma_url=os.environ.get("UPTIME_KUMA_URL", "http://localhost:3001"),
uptime_kuma_api_key=os.environ.get("UPTIME_KUMA_API_KEY", ""),
claude_api_key=os.environ.get("CLAUDE_API_KEY", ""),
uptime_kuma_api_key=resolve_secret("UPTIME_KUMA_API_KEY", ""),
claude_api_key=resolve_secret("CLAUDE_API_KEY", ""),
dev_mode=os.environ.get("DEV_MODE", "false").lower() == "true",
)