import base64
import hashlib
import hmac
import html
import json
import os
import time
from http import cookies
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Optional
from urllib.parse import parse_qs, urlencode, urlparse


PORT = int(os.environ.get("PORT", "8300"))
SECRET = b"unified-local-sso-secret"
USERS = {"admin": "123456"}
TICKET_TTL = 60
TICKETS = {}
ALLOWED_REDIRECTS = {
    "python-app": "http://127.0.0.1:8310/callback",
    "php-app": "http://127.0.0.1:8320/callback",
}


def sign(value: str) -> str:
    return hmac.new(SECRET, value.encode("utf-8"), hashlib.sha256).hexdigest()


def encode_token(payload: dict) -> str:
    raw = base64.urlsafe_b64encode(json.dumps(payload, ensure_ascii=False).encode("utf-8")).decode("utf-8")
    return f"{raw}.{sign(raw)}"


def decode_token(token: str) -> Optional[dict]:
    if "." not in token:
        return None
    raw, signature = token.rsplit(".", 1)
    if not hmac.compare_digest(signature, sign(raw)):
        return None
    try:
        return json.loads(base64.urlsafe_b64decode(raw.encode("utf-8")))
    except Exception:
        return None


def create_ticket(username: str, app_id: str) -> str:
    seed = f"{username}:{app_id}:{time.time()}:{os.urandom(8).hex()}"
    ticket = hashlib.sha256(seed.encode("utf-8")).hexdigest()
    TICKETS[ticket] = {"username": username, "appId": app_id, "expires": time.time() + TICKET_TTL}
    return ticket


def consume_ticket(ticket: str, app_id: str) -> Optional[str]:
    data = TICKETS.pop(ticket, None)
    if not data or data["expires"] < time.time() or data["appId"] != app_id:
        return None
    return data["username"]


class AuthHandler(BaseHTTPRequestHandler):
    def send_html(self, body: str, status: int = 200, headers: Optional[dict] = None) -> None:
        content = body.encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(content)))
        if headers:
            for key, value in headers.items():
                self.send_header(key, value)
        self.end_headers()
        self.wfile.write(content)

    def send_json(self, status: int, payload: dict) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def redirect(self, target: str, headers: Optional[dict] = None) -> None:
        self.send_response(302)
        self.send_header("Location", target)
        if headers:
            for key, value in headers.items():
                self.send_header(key, value)
        self.end_headers()

    def current_user(self) -> Optional[str]:
        jar = cookies.SimpleCookie(self.headers.get("Cookie", ""))
        token = jar.get("sso_login")
        payload = decode_token(token.value) if token else None
        return payload.get("username") if payload else None

    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path == "/":
            self.render_home()
        elif parsed.path == "/login":
            self.render_login(parsed.query)
        elif parsed.path == "/validate":
            self.validate_ticket(parsed.query)
        elif parsed.path == "/logout":
            self.redirect("/", {"Set-Cookie": "sso_login=; Max-Age=0; Path=/; HttpOnly; SameSite=Lax"})
        else:
            self.send_html("not found", 404)

    def do_POST(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path == "/login":
            self.handle_login()
        else:
            self.send_html("not found", 404)

    def render_home(self) -> None:
        user = self.current_user()
        status = f"认证中心已登录：{html.escape(user)}" if user else "认证中心未登录"
        self.send_html(layout("统一 SSO 认证中心", f"""
          <section class="hero"><h1>统一 SSO 认证中心</h1><p>{status}</p><p>这里是唯一登录入口。Python 应用和 PHP 应用都把未登录用户跳转到这里。</p><div class="actions"><a href="http://127.0.0.1:8310/">访问 Python 应用</a><a href="http://127.0.0.1:8320/">访问 PHP 应用</a><a href="/logout">退出认证中心</a></div></section>
          {docs()}
        """))

    def render_login(self, query: str) -> None:
        params = parse_qs(query)
        app_id = params.get("app_id", [""])[0]
        redirect_uri = params.get("redirect_uri", [""])[0]
        if ALLOWED_REDIRECTS.get(app_id) != redirect_uri:
            self.send_html(layout("非法回跳", "<section class='panel'><h1>redirect_uri 不在白名单</h1></section>"), 400)
            return

        user = self.current_user()
        if user:
            ticket = create_ticket(user, app_id)
            self.redirect(f"{redirect_uri}?{urlencode({'ticket': ticket})}")
            return

        self.send_html(layout("统一登录", f"""
          <section class="panel narrow"><h1>统一登录</h1><p>账号：admin，密码：123456。登录成功后认证中心保存统一登录态，并签发当前应用专用 ticket。</p><form method="post" action="/login"><input type="hidden" name="app_id" value="{html.escape(app_id)}"><input type="hidden" name="redirect_uri" value="{html.escape(redirect_uri)}"><label>用户名<input name="username" value="admin"></label><label>密码<input name="password" type="password" value="123456"></label><button type="submit">登录</button></form></section>
        """))

    def handle_login(self) -> None:
        length = int(self.headers.get("Content-Length", "0"))
        data = parse_qs(self.rfile.read(length).decode("utf-8"))
        username = data.get("username", [""])[0]
        password = data.get("password", [""])[0]
        app_id = data.get("app_id", [""])[0]
        redirect_uri = data.get("redirect_uri", [""])[0]

        if ALLOWED_REDIRECTS.get(app_id) != redirect_uri or USERS.get(username) != password:
            self.send_html(layout("登录失败", "<section class='panel'><h1>登录失败</h1><p>应用或账号密码不正确。</p></section>"), 401)
            return

        ticket = create_ticket(username, app_id)
        token = encode_token({"username": username, "iat": int(time.time())})
        self.redirect(f"{redirect_uri}?{urlencode({'ticket': ticket})}", {"Set-Cookie": f"sso_login={token}; Path=/; HttpOnly; SameSite=Lax"})

    def validate_ticket(self, query: str) -> None:
        params = parse_qs(query)
        app_id = params.get("app_id", [""])[0]
        ticket = params.get("ticket", [""])[0]
        username = consume_ticket(ticket, app_id)
        if not username:
            self.send_json(401, {"valid": False, "message": "ticket invalid"})
            return
        self.send_json(200, {"valid": True, "username": username})


def docs() -> str:
    return """
      <section class="grid">
        <article class="panel"><h2>实现了哪些功能</h2><ul><li>统一认证中心</li><li>Python 应用接入 SSO</li><li>PHP 应用接入 SSO</li><li>一次登录，多个应用免重复输入密码</li><li>ticket 一次性校验</li></ul></article>
        <article class="panel"><h2>怎么实现</h2><p>两个业务应用都没有账号密码校验逻辑，只负责检查自己的本地 session。没有 session 时跳转统一认证中心，认证中心已登录就直接签发 ticket 回跳。</p></article>
        <article class="panel"><h2>主要代码段</h2><pre><code>GET /login?app_id=python-app&redirect_uri=...
ticket = create_ticket(user, app_id)
GET /validate?app_id=python-app&ticket=...</code></pre></article>
        <article class="panel"><h2>整体逻辑</h2><ol><li>访问 Python 应用并登录认证中心</li><li>Python 应用拿 ticket 建立本地 session</li><li>访问 PHP 应用时跳到认证中心</li><li>认证中心发现已登录，直接签发 PHP ticket</li><li>PHP 应用建立本地 session</li></ol></article>
        <article class="panel"><h2>注意事项</h2><ul><li>redirect_uri 必须白名单校验</li><li>ticket 必须短期有效且一次性使用</li><li>业务应用只信任认证中心 validate 结果</li><li>真实系统建议使用 OAuth2/OIDC</li></ul></article>
      </section>
    """


def layout(title: str, content: str) -> str:
    return f"""
<!DOCTYPE html><html lang="zh-CN"><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>{html.escape(title)}</title><style>
*{{box-sizing:border-box;margin:0;padding:0}}body{{min-height:100vh;background:#f2f5fb;color:#172033;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI','Microsoft YaHei',sans-serif;line-height:1.7}}.page{{width:min(1120px,calc(100% - 32px));margin:0 auto;padding:34px 0 48px}}.hero,.panel{{border:1px solid #ccd6e6;border-radius:8px;background:#fff;padding:24px}}.hero{{margin-bottom:18px}}h1{{font-size:clamp(30px,5vw,52px);line-height:1.1;letter-spacing:0;margin-bottom:12px}}h2{{font-size:20px;margin-bottom:8px}}p,li{{color:#536176}}.actions{{display:flex;flex-wrap:wrap;gap:10px;margin-top:16px}}a,button{{display:inline-block;border:0;border-radius:6px;background:#2f6fed;color:#fff;padding:10px 16px;font:inherit;font-weight:800;text-decoration:none;cursor:pointer}}.grid{{display:grid;grid-template-columns:repeat(2,minmax(0,1fr));gap:16px}}.narrow{{max-width:520px;margin:60px auto}}label{{display:block;margin-top:12px;color:#394650}}input{{width:100%;margin-top:4px;border:1px solid #c9d1d8;border-radius:6px;padding:10px 12px;font:inherit}}ul,ol{{padding-left:20px}}pre{{overflow-x:auto;border-radius:6px;background:#111827;color:#d7e7ff;padding:14px;font:13px/1.55 Consolas,Monaco,monospace}}@media(max-width:760px){{.grid{{grid-template-columns:1fr}}}}
</style></head><body><main class="page">{content}</main></body></html>
    """


if __name__ == "__main__":
    print(f"Unified SSO auth server running at http://127.0.0.1:{PORT}")
    ThreadingHTTPServer(("127.0.0.1", PORT), AuthHandler).serve_forever()