dekun
2b348854fe
Backups land under /root/eth_hedge_backups; restore accepts zip body for new-server migration and restarts the process. Co-authored-by: Cursor <cursoragent@cursor.com>
84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
"""运行时登录凭据：内存生效 + 持久化到 .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 resolve_env_file_path() -> Path:
|
|
"""第一个已存在的 .env；都不存在则返回仓库根 .env 路径（可写入）。"""
|
|
paths = _env_paths()
|
|
return next((p for p in paths if p.is_file()), paths[0])
|
|
|
|
|
|
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。"""
|
|
if "\n" in value or "\r" in value:
|
|
raise ValueError(f"{key} 值不能包含换行")
|
|
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 ""
|
|
# 简单引号，避免空格/特殊字符破坏解析
|
|
safe = value.replace("\\", "\\\\").replace('"', '\\"')
|
|
line = f'{key}="{safe}"'
|
|
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")
|
|
try:
|
|
target.chmod(0o600)
|
|
except Exception:
|
|
pass
|
|
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()
|