52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
import sqlite3
|
|
import os
|
|
from pathlib import Path
|
|
from fastapi import HTTPException, Header
|
|
|
|
DB_PATH = os.getenv("DB_PATH", "/data/api_keys.db")
|
|
SEED_API_KEY = os.getenv("SEED_API_KEY")
|
|
SEED_GAME = os.getenv("SEED_GAME")
|
|
|
|
def get_db():
|
|
# Ensure the directory exists
|
|
Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
# your-very-long-random-admin-key
|
|
|
|
def init_db():
|
|
conn = get_db()
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS api_keys (
|
|
key TEXT PRIMARY KEY,
|
|
game TEXT NOT NULL,
|
|
active BOOLEAN DEFAULT 1,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
# Insert seed key if provided via env
|
|
if SEED_API_KEY and SEED_GAME:
|
|
try:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO api_keys(key, game) VALUES (?, ?)",
|
|
(SEED_API_KEY, SEED_GAME)
|
|
)
|
|
conn.commit()
|
|
except Exception:
|
|
pass
|
|
conn.close()
|
|
|
|
async def verify_api_key(x_api_key: str = Header(None)) -> str:
|
|
if not x_api_key:
|
|
raise HTTPException(status_code=401, detail="X-API-Key header missing")
|
|
conn = get_db()
|
|
row = conn.execute(
|
|
"SELECT game FROM api_keys WHERE key = ? AND active = 1",
|
|
(x_api_key,)
|
|
).fetchone()
|
|
conn.close()
|
|
if not row:
|
|
raise HTTPException(status_code=403, detail="Invalid or inactive API key")
|
|
return row["game"] |