#!/usr/bin/env python3
"""
auth_proxy.py v2 - Cookie-based session auth proxy for Caddy.
Replaces Caddy basicauth to fix repeated login prompts.
Changes in v2:
- Cookie uses Domain=.xybkwd.top for cross-subdomain sharing
- Removed Secure flag (Caddy terminates TLS, upstream is plain HTTP)
"""
import http.server
import http.client
import urllib.parse
import json
import hmac
import hashlib
import base64
import os
import time
import bcrypt
import re
import mimetypes
# ─── Configuration ─────────────────────────────────────────────
PORT = 9002
SECRET_FILE = '/home/ubuntu/.auth_proxy_secret'
SESSION_DURATION = 7 * 24 * 3600 # 7 days
COOKIE_NAME = 'sess'
NAV_ROOT = '/var/www/nav'
# Password bcrypt hash (fxy / Qiyun2025!)
PASSWORD_HASH = b'$2b$14$0U13bqD7CgmZe7jcBQAUHehV2M6qrotbxERgDrHgzLxSXtmRX/YUK'
# Host -> backend mapping
BACKENDS = {
'gitea.xybkwd.top': ('127.0.0.1', 3000),
'dashboard.xybkwd.top': ('127.0.0.1', 9000),
'books.xybkwd.top': ('127.0.0.1', 8083),
}
STATIC_HOSTS = {
'nav.xybkwd.top', 'brief.xybkwd.top',
'www.xybkwd.top', 'xybkwd.top',
'auth.xybkwd.top',
}
# Load or generate secret key
if os.path.exists(SECRET_FILE):
with open(SECRET_FILE) as f:
SECRET_KEY = f.read().strip()
else:
SECRET_KEY = os.urandom(32).hex()
with open(SECRET_FILE, 'w') as f:
f.write(SECRET_KEY)
os.chmod(SECRET_FILE, 0o600)
# ─── Token management ──────────────────────────────────────────
def make_token(username):
payload = json.dumps({
'u': username,
'e': int(time.time()) + SESSION_DURATION,
'i': os.urandom(4).hex()
}, separators=(',', ':'))
b64 = base64.urlsafe_b64encode(payload.encode()).rstrip(b'=').decode()
sig = hmac.new(SECRET_KEY.encode(), b64.encode(),
hashlib.sha256).hexdigest()[:24]
return b64 + '.' + sig
def verify_token(token):
try:
parts = token.split('.')
if len(parts) != 2:
return None
b64, sig = parts
expected = hmac.new(SECRET_KEY.encode(), b64.encode(),
hashlib.sha256).hexdigest()[:24]
if not hmac.compare_digest(sig, expected):
return None
pad = 4 - len(b64) % 4
if pad != 4:
b64 += '=' * pad
payload = json.loads(base64.urlsafe_b64decode(b64))
if payload.get('e', 0) < time.time():
return None
return payload.get('u')
except Exception:
return None
# ─── Auth handler ──────────────────────────────────────────────
LOGIN_PAGE = '''
栖云 · 登录
'''
class AuthHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
try:
with open('/tmp/auth_proxy.log', 'a') as f:
f.write('{} - {}\n'.format(self.client_address[0], fmt % args))
except Exception:
pass
def get_session_user(self):
cookie = self.headers.get('Cookie', '')
m = re.search(r'{}=([^;]+)'.format(COOKIE_NAME), cookie)
if m:
return verify_token(m.group(1))
return None
def redirect(self, location):
self.send_response(302)
self.send_header('Location', location)
self.end_headers()
def serve_login_page(self, redirect, error=''):
html = LOGIN_PAGE.replace('REDIRECT_PLACEHOLDER', redirect)
if error:
html = html.replace('ERROR_PLACEHOLDER',
'{}
'.format(error))
else:
html = html.replace('ERROR_PLACEHOLDER', '')
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Content-Length', str(len(html.encode())))
self.end_headers()
self.wfile.write(html.encode())
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
params = urllib.parse.parse_qs(parsed.query)
redirect = params.get('redirect', ['https://nav.xybkwd.top/'])[0]
if parsed.path == '/login':
self.serve_login_page(redirect)
elif parsed.path == '/verify':
user = self.get_session_user()
self.send_response(200 if user else 401)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(
json.dumps({'ok': bool(user), 'user': user}).encode())
else:
user = self.get_session_user()
if not user:
actual_url = 'https://{}{}'.format(
self.headers.get('Host', 'nav.xybkwd.top').split(':')[0],
self.path)
login_url = 'https://auth.xybkwd.top/login?redirect={}'.format(
urllib.parse.quote(actual_url))
self.redirect(login_url)
return
host = self.headers.get('Host', 'nav.xybkwd.top').split(':')[0]
if parsed.path.startswith('/api/'):
self.proxy_request('127.0.0.1', 9001)
elif host in STATIC_HOSTS:
self.serve_static(parsed.path)
elif host in BACKENDS:
bh, bp = BACKENDS[host]
self.proxy_request(bh, bp)
else:
self.serve_static(parsed.path)
def do_POST(self):
parsed = urllib.parse.urlparse(self.path)
if parsed.path == '/login':
length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(length).decode('utf-8', errors='replace')
form = urllib.parse.parse_qs(body)
username = form.get('username', [''])[0]
password = form.get('password', [''])[0]
redirect = form.get('redirect', ['https://nav.xybkwd.top/'])[0]
if username == 'fxy' and bcrypt.checkpw(password.encode(), PASSWORD_HASH):
token = make_token(username)
# Domain=.xybkwd.top for cross-subdomain cookie sharing
cookie_val = '{}={}; Path=/; Domain=.xybkwd.top; HttpOnly; SameSite=Lax; Max-Age={}'.format(
COOKIE_NAME, token, SESSION_DURATION)
self.send_response(302)
self.send_header('Location', redirect)
self.send_header('Set-Cookie', cookie_val)
self.end_headers()
else:
params = urllib.parse.parse_qs(parsed.query)
rd = params.get('redirect', ['https://nav.xybkwd.top/'])[0]
self.serve_login_page(rd, '用户名或密码错误')
elif parsed.path == '/verify':
user = self.get_session_user()
self.send_response(200 if user else 401)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(
json.dumps({'ok': bool(user), 'user': user}).encode())
else:
user = self.get_session_user()
if not user:
actual_url = 'https://{}{}'.format(
self.headers.get('Host', 'nav.xybkwd.top').split(':')[0],
self.path)
login_url = 'https://auth.xybkwd.top/login?redirect={}'.format(
urllib.parse.quote(actual_url))
self.redirect(login_url)
return
host = self.headers.get('Host', 'nav.xybkwd.top').split(':')[0]
if parsed.path.startswith('/api/'):
self.proxy_request('127.0.0.1', 9001)
elif host in BACKENDS:
bh, bp = BACKENDS[host]
self.proxy_request(bh, bp)
else:
self.serve_static(parsed.path)
do_PUT = do_POST
do_DELETE = do_POST
do_PATCH = do_POST
# ─── Static file serving ──────────────────────────────────
MIME_MAP = {
'.html': 'text/html; charset=utf-8',
'.css': 'text/css; charset=utf-8',
'.js': 'application/javascript; charset=utf-8',
'.json': 'application/json; charset=utf-8',
'.png': 'image/png',
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.gif': 'image/gif',
'.svg': 'image/svg+xml',
'.ico': 'image/x-icon',
'.webp': 'image/webp',
'.woff2': 'font/woff2',
'.ttf': 'font/ttf',
'.txt': 'text/plain; charset=utf-8',
'.md': 'text/markdown; charset=utf-8',
'.xml': 'text/xml; charset=utf-8',
}
def serve_static(self, path):
if path == '/':
path = '/index.html'
abs_path = os.path.normpath(os.path.join(NAV_ROOT, path.lstrip('/')))
if not abs_path.startswith(os.path.normpath(NAV_ROOT)):
self.send_response(403)
self.end_headers()
self.wfile.write(b'Forbidden')
return
if not os.path.isfile(abs_path):
self.send_response(404)
self.end_headers()
self.wfile.write(b'Not found')
return
ext = os.path.splitext(abs_path)[1].lower()
content_type = self.MIME_MAP.get(ext, 'application/octet-stream')
try:
with open(abs_path, 'rb') as f:
data = f.read()
except IOError:
self.send_response(404)
self.end_headers()
self.wfile.write(b'Not found')
return
self.send_response(200)
self.send_header('Content-Type', content_type)
self.send_header('Content-Length', str(len(data)))
self.send_header('Cache-Control', 'private, max-age=60')
self.end_headers()
self.wfile.write(data)
# ─── Reverse proxy ────────────────────────────────────────
def proxy_request(self, backend_host, backend_port):
try:
conn = http.client.HTTPConnection(backend_host, backend_port, timeout=30)
fwd_headers = {}
for h in ['Host', 'Content-Type', 'Content-Length',
'Accept', 'Accept-Encoding', 'Accept-Language',
'Cookie', 'Referer', 'User-Agent', 'Origin',
'Authorization', 'If-Modified-Since', 'If-None-Match',
'Range', 'X-Forwarded-For', 'X-Real-IP']:
v = self.headers.get(h)
if v:
fwd_headers[h] = v
# Strip our session cookie before forwarding
if 'Cookie' in fwd_headers:
cleaned = re.sub(r'{}=[^;]+;?\s*'.format(COOKIE_NAME), '',
fwd_headers['Cookie']).strip()
if cleaned:
fwd_headers['Cookie'] = cleaned
else:
del fwd_headers['Cookie']
body = None
content_length = self.headers.get('Content-Length')
if content_length:
body = self.rfile.read(int(content_length))
conn.request(self.command, self.path, body=body, headers=fwd_headers)
resp = conn.getresponse()
resp_body = resp.read()
self.send_response(resp.status, resp.reason)
skip_headers = {'transfer-encoding', 'connection', 'keep-alive',
'content-length'}
for h, v in resp.getheaders():
if h.lower() not in skip_headers:
self.send_header(h, v)
self.send_header('Content-Length', str(len(resp_body)))
self.end_headers()
self.wfile.write(resp_body)
conn.close()
except Exception as e:
self.send_response(502)
self.end_headers()
self.wfile.write('Proxy error: {}'.format(e).encode())
def do_HEAD(self):
self.send_response(405)
self.end_headers()
# ─── Main ──────────────────────────────────────────────────────
def main():
server = http.server.HTTPServer(('127.0.0.1', PORT), AuthHandler)
print('Auth proxy listening on 127.0.0.1:{}'.format(PORT))
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
if __name__ == '__main__':
main()