auto-generate token on first run, persist to /data/token.txt, log on startup

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-18 13:44:33 -04:00
parent b44f6866cd
commit 0799595b26

View File

@@ -1,7 +1,9 @@
import logging
import os
import secrets
from pathlib import Path
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
@@ -10,11 +12,51 @@ from .models import CaptureRequest, CaptureResponse, HealthResponse
from .org_writer import format_capture
from .store import IdempotencyStore, get_file_lock
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("synq")
app = FastAPI(title="synq", version="0.1.0")
@asynccontextmanager
async def lifespan(_: FastAPI):
_load_token()
yield
app = FastAPI(title="synq", version="0.1.0", lifespan=lifespan)
_store: IdempotencyStore | None = None
_token_cache: str | None = None
def _load_token() -> str:
global _token_cache
if _token_cache:
return _token_cache
env_token = os.environ.get("PHONE_CAPTURE_TOKEN", "").strip()
if env_token and env_token != "change-me" and env_token != "change-me-long-random-token":
_token_cache = env_token
return _token_cache
token_file = Path(os.environ.get("PHONE_CAPTURE_DB_PATH", "/data/capture.sqlite3")).parent / "token.txt"
if token_file.exists():
stored = token_file.read_text().strip()
if stored:
_token_cache = stored
logger.info("synq token loaded from %s", token_file)
return _token_cache
generated = secrets.token_hex(32)
token_file.parent.mkdir(parents=True, exist_ok=True)
token_file.write_text(generated)
logger.info("")
logger.info("=" * 60)
logger.info(" SYNQ TOKEN (paste into app settings):")
logger.info(" %s", generated)
logger.info("=" * 60)
logger.info("")
_token_cache = generated
return _token_cache
def get_store() -> IdempotencyStore:
@@ -30,7 +72,7 @@ def _org_path() -> str:
def _token() -> str:
return os.environ.get("PHONE_CAPTURE_TOKEN", "")
return _load_token()
async def check_token(request: Request) -> None: