Remove API URL UI; add username/password change in settings.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -3,18 +3,25 @@ from __future__ import annotations
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from ..config import Settings, get_settings
|
||||
from ..credentials import get_credentials, update_credentials
|
||||
from .auth import LoginRequest, LoginResponse, issue_token, require_user
|
||||
|
||||
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||||
|
||||
|
||||
class ChangeCredentialsRequest(BaseModel):
|
||||
current_password: str = Field(min_length=1)
|
||||
new_username: str = Field(min_length=1, max_length=64)
|
||||
new_password: str = Field(min_length=4, max_length=128)
|
||||
|
||||
|
||||
@router.post("/login", response_model=LoginResponse)
|
||||
async def login(body: LoginRequest, settings: Annotated[Settings, Depends(get_settings)]) -> LoginResponse:
|
||||
user_ok = body.username == settings.auth_username
|
||||
pass_ok = body.password == settings.auth_password
|
||||
if not (user_ok and pass_ok):
|
||||
user, pwd = get_credentials()
|
||||
if body.username != user or body.password != pwd:
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误")
|
||||
token, ttl = issue_token(body.username, settings)
|
||||
return LoginResponse(
|
||||
@@ -34,3 +41,26 @@ async def me(username: Annotated[str, Depends(require_user)], settings: Annotate
|
||||
"mode": settings.mode,
|
||||
"sim": settings.is_sim,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/change-credentials", response_model=LoginResponse)
|
||||
async def change_credentials(
|
||||
body: ChangeCredentialsRequest,
|
||||
username: Annotated[str, Depends(require_user)],
|
||||
settings: Annotated[Settings, Depends(get_settings)],
|
||||
) -> LoginResponse:
|
||||
_cur_user, cur_pwd = get_credentials()
|
||||
if body.current_password != cur_pwd:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="当前密码不正确")
|
||||
try:
|
||||
update_credentials(new_username=body.new_username, new_password=body.new_password)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
|
||||
token, ttl = issue_token(body.new_username.strip(), settings)
|
||||
return LoginResponse(
|
||||
token=token,
|
||||
username=body.new_username.strip(),
|
||||
expires_in=ttl,
|
||||
env_name=settings.env_name,
|
||||
mode=settings.mode,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
"""运行时登录凭据:内存生效 + 持久化到 .env。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from threading import Lock
|
||||
|
||||
from .config import get_settings
|
||||
|
||||
_lock = Lock()
|
||||
_username: str | None = None
|
||||
_password: str | None = None
|
||||
|
||||
|
||||
def _env_paths() -> list[Path]:
|
||||
here = Path(__file__).resolve()
|
||||
# backend/app/credentials.py -> repo root = parents[2]
|
||||
root = here.parents[2]
|
||||
return [root / ".env", Path.cwd() / ".env", Path.cwd().parent / ".env"]
|
||||
|
||||
|
||||
def _ensure_loaded() -> None:
|
||||
global _username, _password
|
||||
if _username is not None and _password is not None:
|
||||
return
|
||||
s = get_settings()
|
||||
_username = s.auth_username
|
||||
_password = s.auth_password
|
||||
|
||||
|
||||
def get_credentials() -> tuple[str, str]:
|
||||
with _lock:
|
||||
_ensure_loaded()
|
||||
assert _username is not None and _password is not None
|
||||
return _username, _password
|
||||
|
||||
|
||||
def upsert_env_file(key: str, value: str) -> Path | None:
|
||||
"""写入第一个已存在的 .env;都不存在则写仓库根 .env。"""
|
||||
paths = _env_paths()
|
||||
target = next((p for p in paths if p.is_file()), paths[0])
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
text = target.read_text(encoding="utf-8") if target.is_file() else ""
|
||||
line = f"{key}={value}"
|
||||
pattern = re.compile(rf"(?m)^{re.escape(key)}=.*$")
|
||||
if pattern.search(text):
|
||||
text = pattern.sub(line, text)
|
||||
else:
|
||||
if text and not text.endswith("\n"):
|
||||
text += "\n"
|
||||
text += line + "\n"
|
||||
target.write_text(text, encoding="utf-8")
|
||||
return target
|
||||
|
||||
|
||||
def update_credentials(*, new_username: str, new_password: str) -> None:
|
||||
global _username, _password
|
||||
user = new_username.strip()
|
||||
pwd = new_password
|
||||
if not user or not pwd:
|
||||
raise ValueError("用户名和密码不能为空")
|
||||
with _lock:
|
||||
upsert_env_file("AUTH_USERNAME", user)
|
||||
upsert_env_file("AUTH_PASSWORD", pwd)
|
||||
_username = user
|
||||
_password = pwd
|
||||
# 刷新 Settings 缓存,避免进程内读到旧值
|
||||
get_settings.cache_clear()
|
||||
Reference in New Issue
Block a user