370 lines
14 KiB
Python
Executable File
370 lines
14 KiB
Python
Executable File
#!/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
|
|
- Added Secure flag (Caddy terminates TLS, browser sees HTTPS, cookie needs Secure)
|
|
"""
|
|
|
|
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 = '''<!DOCTYPE html>
|
|
<html lang="zh-CN">
|
|
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1.0">
|
|
<title>栖云 · 登录</title>
|
|
<style>
|
|
*{margin:0;padding:0;box-sizing:border-box}
|
|
body{font-family:-apple-system,BlinkMacSystemFont,"SF Pro Display","Helvetica Neue","PingFang SC",sans-serif;background:#13131a;color:#f0f0f5;min-height:100vh;display:flex;align-items:center;justify-content:center;padding:20px}
|
|
.card{background:rgba(40,40,50,0.85);backdrop-filter:blur(20px);border-radius:20px;padding:40px;width:100%;max-width:340px;box-shadow:0 8px 32px rgba(0,0,0,0.3);border:1px solid rgba(255,255,255,0.06)}
|
|
h1{font-size:22px;font-weight:600;margin-bottom:4px;text-align:center}
|
|
.sub{color:#98989d;font-size:13px;text-align:center;margin-bottom:28px}
|
|
.form-group{margin-bottom:16px}
|
|
label{display:block;font-size:12px;color:#98989d;margin-bottom:4px}
|
|
input{width:100%;padding:12px 14px;border-radius:12px;border:1px solid rgba(255,255,255,0.1);background:rgba(255,255,255,0.05);color:#f0f0f5;font-size:15px;outline:none;transition:border .2s}
|
|
input:focus{border-color:#7c7cf0}
|
|
button{width:100%;padding:12px;border-radius:12px;border:none;background:#7c7cf0;color:#fff;font-size:15px;font-weight:600;cursor:pointer;margin-top:4px}
|
|
button:hover{background:#6a6ae0}
|
|
.error{color:#ff3b30;font-size:12px;text-align:center;margin-top:12px}
|
|
.footer{font-size:11px;color:#636366;text-align:center;margin-top:20px}
|
|
</style></head>
|
|
<body><div class="card">
|
|
<h1>栖云</h1>
|
|
<p class="sub">个人数据中心</p>
|
|
<form method="post" action="/login">
|
|
<input type="hidden" name="redirect" value="REDIRECT_PLACEHOLDER">
|
|
<div class="form-group"><label>用户名</label><input type="text" name="username" placeholder="fxy" autocomplete="username"></div>
|
|
<div class="form-group"><label>密码</label><input type="password" name="password" placeholder="输入密码" autocomplete="current-password"></div>
|
|
<button type="submit">登录</button>
|
|
ERROR_PLACEHOLDER
|
|
</form><div class="footer">auth.xybkwd.top</div>
|
|
</div></body></html>'''
|
|
|
|
|
|
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',
|
|
'<div class="error">{}</div>'.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()
|