dekun
150 lines
4.5 KiB
Python
150 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import hmac
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from ..auth import issue_token, require_control_user
|
|
from ..config import ControlSettings, get_control_settings
|
|
from ..envfile import update_control_credentials, upsert_env_control
|
|
from ..lan import client_ip, is_lan_ip
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
|
|
class LoginBody(BaseModel):
|
|
username: str = Field(min_length=1)
|
|
password: str = Field(min_length=1)
|
|
|
|
|
|
class ChangeCredentialsBody(BaseModel):
|
|
current_password: str = Field(min_length=1)
|
|
new_username: str = Field(min_length=1, max_length=64)
|
|
new_password: str = Field(min_length=6, max_length=128)
|
|
|
|
|
|
class LanBypassBody(BaseModel):
|
|
enabled: bool
|
|
|
|
|
|
@router.get("/login-meta")
|
|
async def login_meta(
|
|
request: Request,
|
|
settings: Annotated[ControlSettings, Depends(get_control_settings)],
|
|
) -> dict:
|
|
ip = client_ip(request)
|
|
lan = is_lan_ip(ip)
|
|
return {
|
|
"show_default_hint": settings.is_default_credentials,
|
|
"lan_bypass_enabled": settings.lan_auth_bypass,
|
|
"lan_client": lan,
|
|
"lan_login_available": bool(settings.lan_auth_bypass and lan),
|
|
"client_ip": ip or None,
|
|
}
|
|
|
|
|
|
@router.post("/lan-login")
|
|
async def lan_login(
|
|
request: Request,
|
|
settings: Annotated[ControlSettings, Depends(get_control_settings)],
|
|
) -> dict:
|
|
if not settings.lan_auth_bypass:
|
|
raise HTTPException(status_code=403, detail="未开启局域网免登录")
|
|
ip = client_ip(request)
|
|
if not is_lan_ip(ip):
|
|
raise HTTPException(status_code=403, detail="仅局域网地址可免登录")
|
|
user = settings.control_auth_username
|
|
token, ttl = issue_token(user, settings)
|
|
return {
|
|
"token": token,
|
|
"username": user,
|
|
"expires_in": ttl,
|
|
"via": "lan",
|
|
}
|
|
|
|
|
|
@router.post("/login")
|
|
async def login(
|
|
body: LoginBody,
|
|
settings: Annotated[ControlSettings, Depends(get_control_settings)],
|
|
) -> dict:
|
|
user_ok = hmac.compare_digest(
|
|
body.username.encode("utf-8"),
|
|
settings.control_auth_username.encode("utf-8"),
|
|
)
|
|
pwd_ok = hmac.compare_digest(
|
|
body.password.encode("utf-8"),
|
|
settings.control_auth_password.encode("utf-8"),
|
|
)
|
|
if not (user_ok and pwd_ok):
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
token, ttl = issue_token(body.username, settings)
|
|
return {"token": token, "username": body.username, "expires_in": ttl}
|
|
|
|
|
|
@router.get("/me")
|
|
async def me(
|
|
request: Request,
|
|
username: Annotated[str, Depends(require_control_user)],
|
|
settings: Annotated[ControlSettings, Depends(get_control_settings)],
|
|
) -> dict:
|
|
ip = client_ip(request)
|
|
lan = is_lan_ip(ip)
|
|
return {
|
|
"username": username,
|
|
"poll_interval_sec": settings.control_poll_interval_sec,
|
|
"sse_interval_sec": settings.control_sse_interval_sec,
|
|
"show_default_hint": settings.is_default_credentials,
|
|
"lan_bypass_enabled": settings.lan_auth_bypass,
|
|
"lan_client": lan,
|
|
"client_ip": ip or None,
|
|
}
|
|
|
|
|
|
@router.put("/lan-bypass")
|
|
async def put_lan_bypass(
|
|
body: LanBypassBody,
|
|
_user: Annotated[str, Depends(require_control_user)],
|
|
) -> dict:
|
|
upsert_env_control(
|
|
"CONTROL_LAN_AUTH_BYPASS",
|
|
"1" if body.enabled else "0",
|
|
overwrite=True,
|
|
)
|
|
get_control_settings.cache_clear()
|
|
settings = get_control_settings()
|
|
return {
|
|
"ok": True,
|
|
"lan_bypass_enabled": settings.lan_auth_bypass,
|
|
}
|
|
|
|
|
|
@router.post("/change-credentials")
|
|
async def change_credentials(
|
|
body: ChangeCredentialsBody,
|
|
username: Annotated[str, Depends(require_control_user)],
|
|
settings: Annotated[ControlSettings, Depends(get_control_settings)],
|
|
) -> dict:
|
|
if not hmac.compare_digest(
|
|
body.current_password.encode("utf-8"),
|
|
settings.control_auth_password.encode("utf-8"),
|
|
):
|
|
raise HTTPException(status_code=400, detail="当前密码不正确")
|
|
try:
|
|
update_control_credentials(
|
|
new_username=body.new_username,
|
|
new_password=body.new_password,
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e)) from e
|
|
settings2 = get_control_settings()
|
|
token, ttl = issue_token(body.new_username.strip(), settings2)
|
|
return {
|
|
"token": token,
|
|
"username": body.new_username.strip(),
|
|
"expires_in": ttl,
|
|
"show_default_hint": settings2.is_default_credentials,
|
|
}
|