Your Name
115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
清除登录防暴破锁定（删除 login_attempts 中近期失败记录）。
|
||
|
||
锁定规则见 server/application/services/auth_service.py：
|
||
MAX_LOGIN_FAILURES=5，LOCKOUT_MINUTES=15（按 username + ip_address 统计）
|
||
|
||
用法:
|
||
python scripts/unlock_login_lockout.py
|
||
python scripts/unlock_login_lockout.py --username admin
|
||
python scripts/unlock_login_lockout.py --all-failures # 删除该用户全部失败记录
|
||
|
||
需在仓库根目录有 .env（含 NEXUS_DATABASE_URL 或 DATABASE_URL），或在生产机部署目录执行。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import os
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from urllib.parse import unquote, urlparse
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
def _load_dotenv() -> None:
|
||
env_path = ROOT / ".env"
|
||
if not env_path.is_file():
|
||
return
|
||
for line in env_path.read_text(encoding="utf-8", errors="replace").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, value = line.partition("=")
|
||
key, value = key.strip(), value.strip().strip("\"'")
|
||
if key and key not in os.environ:
|
||
os.environ[key] = value
|
||
|
||
|
||
def _parse_database_url(url: str) -> dict:
|
||
"""Parse mysql+aiomysql://user:pass@host:port/db"""
|
||
u = urlparse(url.replace("mysql+aiomysql://", "mysql://", 1))
|
||
return {
|
||
"host": u.hostname or "127.0.0.1",
|
||
"port": u.port or 3306,
|
||
"user": unquote(u.username or ""),
|
||
"password": unquote(u.password or ""),
|
||
"db": (u.path or "/").lstrip("/") or "nexus",
|
||
}
|
||
|
||
|
||
async def _unlock(username: str, all_failures: bool) -> int:
|
||
import aiomysql
|
||
|
||
raw = (
|
||
os.environ.get("NEXUS_DATABASE_URL")
|
||
or os.environ.get("DATABASE_URL")
|
||
or ""
|
||
)
|
||
if not raw:
|
||
print("ERROR: 未找到 NEXUS_DATABASE_URL / DATABASE_URL，请在 .env 中配置或在生产目录执行", file=sys.stderr)
|
||
return 1
|
||
|
||
cfg = _parse_database_url(raw)
|
||
conn = await aiomysql.connect(
|
||
host=cfg["host"],
|
||
port=int(cfg["port"]),
|
||
user=cfg["user"],
|
||
password=cfg["password"],
|
||
db=cfg["db"],
|
||
connect_timeout=20,
|
||
)
|
||
try:
|
||
async with conn.cursor() as cur:
|
||
if all_failures:
|
||
await cur.execute(
|
||
"DELETE FROM login_attempts WHERE username = %s AND success = 0",
|
||
(username,),
|
||
)
|
||
else:
|
||
await cur.execute(
|
||
"""
|
||
DELETE FROM login_attempts
|
||
WHERE username = %s AND success = 0
|
||
AND attempted_at >= UTC_TIMESTAMP() - INTERVAL 20 MINUTE
|
||
""",
|
||
(username,),
|
||
)
|
||
deleted = cur.rowcount
|
||
await conn.commit()
|
||
print(f"OK: 已删除用户「{username}」失败登录记录 {deleted} 条，锁定已解除。")
|
||
return 0
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="解除 Nexus 登录防暴破锁定")
|
||
parser.add_argument("--username", default="admin", help="管理员用户名（默认 admin）")
|
||
parser.add_argument(
|
||
"--all-failures",
|
||
action="store_true",
|
||
help="删除该用户全部历史失败记录（默认仅近 20 分钟）",
|
||
)
|
||
args = parser.parse_args()
|
||
_load_dotenv()
|
||
return asyncio.run(_unlock(args.username, args.all_failures))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|