Control settings tabs, hide default creds hint after change, LAN passwordless login.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -3,12 +3,13 @@ from __future__ import annotations
|
||||
import hmac
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
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
|
||||
from ..envfile import update_control_credentials, upsert_env_control
|
||||
from ..lan import client_ip, is_lan_ip
|
||||
|
||||
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||||
|
||||
@@ -24,6 +25,46 @@ class ChangeCredentialsBody(BaseModel):
|
||||
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,
|
||||
@@ -51,6 +92,26 @@ async def me(
|
||||
return {
|
||||
"username": username,
|
||||
"poll_interval_sec": settings.control_poll_interval_sec,
|
||||
"show_default_hint": settings.is_default_credentials,
|
||||
"lan_bypass_enabled": settings.lan_auth_bypass,
|
||||
}
|
||||
|
||||
|
||||
@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,
|
||||
}
|
||||
|
||||
|
||||
@@ -78,4 +139,5 @@ async def change_credentials(
|
||||
"token": token,
|
||||
"username": body.new_username.strip(),
|
||||
"expires_in": ttl,
|
||||
"show_default_hint": settings2.is_default_credentials,
|
||||
}
|
||||
|
||||
@@ -33,6 +33,8 @@ class ControlSettings(BaseSettings):
|
||||
control_poll_interval_sec: int = 8
|
||||
control_http_timeout_sec: float = 12.0
|
||||
control_port: int = 5160
|
||||
# "1"/"0":局域网客户端免密登录
|
||||
control_lan_auth_bypass: str = "0"
|
||||
|
||||
@property
|
||||
def db_path(self) -> Path:
|
||||
@@ -40,6 +42,22 @@ class ControlSettings(BaseSettings):
|
||||
return Path(self.control_db_path)
|
||||
return _control_root() / "data" / "control.db"
|
||||
|
||||
@property
|
||||
def lan_auth_bypass(self) -> bool:
|
||||
return self.control_lan_auth_bypass.strip().lower() in (
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
"on",
|
||||
)
|
||||
|
||||
@property
|
||||
def is_default_credentials(self) -> bool:
|
||||
return (
|
||||
self.control_auth_username.strip() == "admin"
|
||||
and self.control_auth_password == "admin123"
|
||||
)
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_control_settings() -> ControlSettings:
|
||||
|
||||
@@ -73,6 +73,7 @@ _DEPLOY_DEFAULTS: dict[str, str] = {
|
||||
"CONTROL_POLL_INTERVAL_SEC": "8",
|
||||
"CONTROL_HTTP_TIMEOUT_SEC": "12",
|
||||
"CONTROL_AUTH_TOKEN_VERSION": "1",
|
||||
"CONTROL_LAN_AUTH_BYPASS": "0",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
"""客户端 IP / 局域网判断。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
|
||||
from fastapi import Request
|
||||
|
||||
|
||||
def client_ip(request: Request) -> str:
|
||||
xff = request.headers.get("x-forwarded-for") or ""
|
||||
if xff.strip():
|
||||
return xff.split(",")[0].strip()
|
||||
if request.client and request.client.host:
|
||||
return request.client.host
|
||||
return ""
|
||||
|
||||
|
||||
def is_lan_ip(ip: str) -> bool:
|
||||
raw = (ip or "").strip()
|
||||
if not raw:
|
||||
return False
|
||||
if raw in ("localhost", "::1"):
|
||||
return True
|
||||
try:
|
||||
addr = ipaddress.ip_address(raw.split("%")[0])
|
||||
except ValueError:
|
||||
return False
|
||||
return bool(addr.is_loopback or addr.is_private)
|
||||
Reference in New Issue
Block a user