jason/qrknit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ops: async click pipeline, honest uptime, backups, retention; auth/audit plumbing

Browse Source 
- Click events are enqueued and written by a background thread — redirects
  never block on DB writes or geo lookups.
- Uptime heartbeat marks per-minute rows (INSERT OR IGNORE), safe across
  Gunicorn workers; /api/uptime now measures real availability from each
  day's first heartbeat instead of a per-process counter that could only
  report 100%. Legacy uptime_daily rows still honoured.
- Hourly maintenance thread: daily SQLite backup via VACUUM INTO (atomic
  temp-file rename, race-safe across workers, BACKUP_KEEP retention),
  CLICK_RETENTION_DAYS purge, uptime-mark pruning.
- notify_webhook() helper for Slack/Discord-compatible WEBHOOK_URL.
- Auth groundwork: _try_api_key_auth() lets the login/admin decorators
  accept Bearer qrk_ tokens (scope-checked, no Set-Cookie), and
  log_audit() records admin actions — used by the following feature
  commits.
- New env config: WEBHOOK_URL, GEOIP_DB_PATH, GEO_API_FALLBACK,
  IP_ANONYMIZE, CLICK_RETENTION_DAYS, BACKUPS, BACKUP_DIR, BACKUP_KEEP.
- requirements: add geoip2 for local GeoLite2 lookups.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jason Stedwell
2026-07-03 01:39:50 -05:00
parent c892588d31
commit d0e99b5244
2 changed files with 202 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+201 -18
View File
@@ -17,6 +17,9 @@ import hashlib
import hmac
import time
import io
import json
import queue
import secrets
import struct
import zlib
import threading
@@ -40,6 +43,16 @@ DB_PATH = os.environ.get('DB_PATH', '/app/data/qrknit.db')
BASE_URL = os.environ.get('BASE_URL', 'http://localhost:5000').rstrip('/')
APP_NAME = os.environ.get('APP_NAME', 'to.ALWISP')
# Operational settings — see INSTALL.md
WEBHOOK_URL = os.environ.get('WEBHOOK_URL', '') # Slack/Discord-compatible webhook for admin notifications
GEOIP_DB_PATH = os.environ.get('GEOIP_DB_PATH', '/app/data/GeoLite2-Country.mmdb')
GEO_API_FALLBACK = os.environ.get('GEO_API_FALLBACK', 'false').lower() == 'true' # ip-api.com is non-commercial-use only
IP_ANONYMIZE = os.environ.get('IP_ANONYMIZE', 'none').lower() # none | truncate | hash
CLICK_RETENTION_DAYS = int(os.environ.get('CLICK_RETENTION_DAYS', '0')) # 0 = keep click events forever
BACKUPS_ENABLED = os.environ.get('BACKUPS', 'true').lower() == 'true'
BACKUP_DIR = os.environ.get('BACKUP_DIR', '') # defaults to <db dir>/backups
BACKUP_KEEP = int(os.environ.get('BACKUP_KEEP', '7'))
COOKIE_SECURE = os.environ.get('COOKIE_SECURE', 'false').lower() == 'true'
app.config['SESSION_COOKIE_HTTPONLY'] = True
app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
@@ -212,19 +225,22 @@ init_db()
seed_admin()
def now_iso():
return datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
def _uptime_heartbeat():
"""Daemon thread: records a 'checks_up' ping every 60 s to track daily availability."""
"""Daemon thread: marks the current UTC minute as 'up' once per minute.
INSERT OR IGNORE on the minute primary key makes this safe to run in every
Gunicorn worker — multiple workers collapse to a single mark per minute,
so uptime is no longer over-counted in multi-worker deployments.
"""
while True:
try:
today = datetime.now(timezone.utc).date().isoformat()
minute = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M')
with get_db() as conn:
conn.execute("""
INSERT INTO uptime_daily (date, checks_total, checks_up)
VALUES (?, 1, 1)
ON CONFLICT(date) DO UPDATE SET
checks_total = checks_total + 1,
checks_up = checks_up + 1
""", (today,))
conn.execute('INSERT OR IGNORE INTO uptime_minutes (minute) VALUES (?)', (minute,))
except Exception:
pass
time.sleep(60)
@@ -233,12 +249,146 @@ threading.Thread(target=_uptime_heartbeat, daemon=True).start()
# ─────────────────────────────────────────────
# Auth decorators
# Async click recording
# Clicks are enqueued from the redirect path and written by a background
# thread, so redirects never block on DB writes or geo lookups.
# ─────────────────────────────────────────────
_click_queue: "queue.Queue" = queue.Queue(maxsize=10000)
def _click_writer():
while True:
item = _click_queue.get()
try:
country = item['cf_country'] or get_country_for_ip(item['ip'])
with get_db() as conn:
conn.execute(
'INSERT INTO clicks (link_id,clicked_at,referrer,user_agent,ip_address,country,is_bot) '
'VALUES (?,?,?,?,?,?,?)',
(item['link_id'], item['ts'], item['referrer'], item['ua'],
anonymize_ip(item['ip']), country, item['is_bot'])
)
conn.execute('UPDATE links SET clicks=clicks+1 WHERE id=?', (item['link_id'],))
except Exception:
pass
threading.Thread(target=_click_writer, daemon=True).start()
# ─────────────────────────────────────────────
# Daily maintenance — backups, retention, pruning
# ─────────────────────────────────────────────
def _run_maintenance():
today = datetime.now(timezone.utc).date().isoformat()
cutoff = (datetime.now(timezone.utc) - timedelta(days=31)).strftime('%Y-%m-%dT%H:%M')
with get_db() as conn:
conn.execute('DELETE FROM uptime_minutes WHERE minute < ?', (cutoff,))
if CLICK_RETENTION_DAYS > 0:
click_cutoff = (datetime.now(timezone.utc).replace(tzinfo=None)
- timedelta(days=CLICK_RETENTION_DAYS)).isoformat()
conn.execute('DELETE FROM clicks WHERE clicked_at < ?', (click_cutoff,))
if BACKUPS_ENABLED:
backup_dir = BACKUP_DIR or os.path.join(os.path.dirname(DB_PATH), 'backups')
os.makedirs(backup_dir, exist_ok=True)
path = os.path.join(backup_dir, f'qrknit-{today}.db')
if not os.path.exists(path):
# Vacuum into a per-process temp file, then rename atomically —
# concurrent workers can't trip over a half-written target.
tmp = f'{path}.tmp{os.getpid()}'
try:
conn = sqlite3.connect(DB_PATH)
try:
conn.execute('VACUUM INTO ?', (tmp,))
finally:
conn.close()
os.replace(tmp, path)
except Exception:
try:
os.remove(tmp)
except OSError:
pass
backups = sorted(f for f in os.listdir(backup_dir)
if f.startswith('qrknit-') and f.endswith('.db'))
for old in backups[:-BACKUP_KEEP] if BACKUP_KEEP > 0 else []:
try:
os.remove(os.path.join(backup_dir, old))
except OSError:
pass
def _maintenance_loop():
while True:
try:
_run_maintenance()
except Exception:
pass
time.sleep(3600)
threading.Thread(target=_maintenance_loop, daemon=True).start()
# ─────────────────────────────────────────────
# Webhook notifications (Slack / Discord compatible)
# ─────────────────────────────────────────────
def notify_webhook(text):
"""POST a notification to WEBHOOK_URL. Fire-and-forget — never blocks a request."""
if not WEBHOOK_URL:
return
def _send():
try:
import urllib.request as _ureq
payload = json.dumps({'text': text, 'content': text}).encode()
req = _ureq.Request(WEBHOOK_URL, data=payload,
headers={'Content-Type': 'application/json', 'User-Agent': 'QRknit/1.0'})
_ureq.urlopen(req, timeout=5)
except Exception:
pass
threading.Thread(target=_send, daemon=True).start()
# ─────────────────────────────────────────────
# Auth decorators
# Requests authenticate with either a session cookie or an API key
# (`Authorization: Bearer qrk_…`). Keys carry a read or write scope.
# ─────────────────────────────────────────────
def _try_api_key_auth():
"""If a Bearer API key is present, validate it and load the key's user into
the request's session (without emitting a Set-Cookie). Returns an error
response tuple on a bad key / insufficient scope, else None."""
auth = request.headers.get('Authorization', '')
if not auth.startswith('Bearer '):
return None
token = auth[7:].strip()
if not token.startswith('qrk_'):
return jsonify({'error': 'Invalid API key'}), 401
key_hash = hashlib.sha256(token.encode()).hexdigest()
with get_db() as conn:
row = conn.execute(
'SELECT k.id, k.scope, u.id AS user_id, u.username, u.is_admin, u.org_id '
'FROM api_keys k JOIN users u ON k.user_id=u.id '
'WHERE k.key_hash=? AND k.revoked=0', (key_hash,)
).fetchone()
if not row:
return jsonify({'error': 'Invalid API key'}), 401
if row['scope'] != 'write' and request.method not in ('GET', 'HEAD', 'OPTIONS'):
return jsonify({'error': 'This API key is read-only'}), 403
conn.execute('UPDATE api_keys SET last_used_at=? WHERE id=?', (now_iso(), row['id']))
session['authenticated'] = True
session['user_id'] = row['user_id']
session['username'] = row['username']
session['is_admin'] = bool(row['is_admin'])
session['org_id'] = row['org_id']
session.modified = False # request-scoped only — no session cookie for API clients
return None
def login_required(f):
@wraps(f)
def decorated(*args, **kwargs):
err = _try_api_key_auth()
if err:
return err
if not session.get('authenticated') or not session.get('user_id'):
return jsonify({'error': 'Authentication required'}), 401
return f(*args, **kwargs)
@@ -247,6 +397,9 @@ def login_required(f):
def admin_required(f):
@wraps(f)
def decorated(*args, **kwargs):
err = _try_api_key_auth()
if err:
return err
if not session.get('authenticated') or not session.get('user_id'):
return jsonify({'error': 'Authentication required'}), 401
if not session.get('is_admin'):
@@ -255,6 +408,14 @@ def admin_required(f):
return decorated
def log_audit(conn, action, target=None, details=None):
"""Record an administrative action in the audit log."""
conn.execute(
'INSERT INTO audit_log (actor, action, target, details, created_at) VALUES (?,?,?,?,?)',
(session.get('username'), action, target, details, now_iso())
)
# ─────────────────────────────────────────────
# Plan / usage helpers
# ─────────────────────────────────────────────
@@ -577,21 +738,43 @@ def health():
@app.route('/api/uptime')
def get_uptime():
"""Public endpoint — 30-day daily uptime history for the header SLA graph."""
today = datetime.now(timezone.utc).date()
"""Public endpoint — 30-day daily uptime history for the header SLA graph.
Computed from per-minute 'up' marks: pct = minutes marked / minutes elapsed
that day, so downtime actually shows up (the old per-process counter could
only ever report 100%, multiplied per worker). Falls back to legacy
uptime_daily rows for days that predate the minute marks.
"""
now = datetime.now(timezone.utc)
today = now.date()
start = today - timedelta(days=29)
with get_db() as conn:
rows = conn.execute(
"SELECT date, checks_total, checks_up FROM uptime_daily WHERE date >= ? ORDER BY date",
"SELECT substr(minute,1,10) as day, COUNT(*) as mins, MIN(minute) as first "
"FROM uptime_minutes WHERE minute >= ? GROUP BY day",
(start.isoformat(),)
).fetchall()
row_map = {r['date']: r for r in rows}
legacy = conn.execute(
"SELECT date, checks_total, checks_up FROM uptime_daily WHERE date >= ?",
(start.isoformat(),)
).fetchall()
min_map = {r['day']: r for r in rows}
legacy_map = {r['date']: r for r in legacy}
days = []
for i in range(30):
d = (start + timedelta(days=i)).isoformat()
row = row_map.get(d)
if row and row['checks_total'] > 0:
pct = round(row['checks_up'] / row['checks_total'] * 100, 2)
d_date = start + timedelta(days=i)
d = d_date.isoformat()
row = min_map.get(d)
if row:
# Window runs from the day's first heartbeat (so fresh installs and
# mid-day deployments aren't penalised) to end-of-day / now.
first_min = int(row['first'][11:13]) * 60 + int(row['first'][14:16])
end_min = 1439 if d_date < today else now.hour * 60 + now.minute
expected = max(1, end_min - first_min + 1)
pct = round(min(row['mins'] / expected, 1.0) * 100, 2)
elif d in legacy_map and legacy_map[d]['checks_total'] > 0:
r = legacy_map[d]
pct = round(r['checks_up'] / r['checks_total'] * 100, 2)
else:
pct = None
days.append({'date': d, 'pct': pct})
requirements.txt
+1
View File
@@ -2,3 +2,4 @@ flask>=2.3.0
qrcode[pil]>=7.4.2
Pillow>=10.0.0
gunicorn>=21.0.0
geoip2>=4.7.0