78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
"""直接用 paramiko + 容器内 Python 重置 owner 密码为固定值,然后验证登录。"""
|
|
import os, paramiko, json
|
|
PW = os.environ["REMOTE_PASS"]
|
|
NEW_PW = "Owner2026!"
|
|
|
|
c = paramiko.SSHClient()
|
|
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
c.connect("207.57.129.228", port=19717, username="root", password=PW, timeout=15, allow_agent=False, look_for_keys=False)
|
|
def run(cmd, t=30):
|
|
si, so, se = c.exec_command(cmd, timeout=t)
|
|
out = so.read().decode("utf-8", "replace")
|
|
err = se.read().decode("utf-8", "replace")
|
|
rc = so.channel.recv_exit_status()
|
|
if out: print(out, end="")
|
|
if err: print("[err]", err, end="", file=__import__("sys").stderr)
|
|
return out
|
|
|
|
# 在 api 容器内用 python 算 bcrypt hash + update
|
|
print("--- 重设密码 ---")
|
|
quoted = NEW_PW.replace('"', '\\"')
|
|
cmd = f'''docker exec news-aggregator-api-1 python -c "
|
|
from app.core.security import hash_password
|
|
from app.database import AsyncSessionLocal
|
|
from app.models.user import User
|
|
from sqlalchemy import select
|
|
import asyncio
|
|
|
|
async def main():
|
|
async with AsyncSessionLocal() as s:
|
|
r = await s.execute(select(User).where(User.username == 'owner'))
|
|
u = r.scalar_one_or_none()
|
|
if u is None:
|
|
print('NO USER')
|
|
return
|
|
u.password_hash = hash_password('{NEW_PW}')
|
|
u.role = 'owner'
|
|
await s.commit()
|
|
print('OK', u.id, u.username, u.role.value)
|
|
|
|
asyncio.run(main())
|
|
"'''
|
|
out = run(cmd, t=30)
|
|
print(out)
|
|
|
|
# 写文件
|
|
run(f'echo "{NEW_PW}" > /root/.owner_pass && chmod 600 /root/.owner_pass')
|
|
print(f" /root/.owner_pass = {NEW_PW}")
|
|
|
|
# 登录
|
|
print("\n--- 登录 ---")
|
|
import urllib.parse
|
|
body = json.dumps({"username": "owner", "password": NEW_PW})
|
|
out = run(f"curl -s -X POST http://localhost/api/v1/auth/login -H 'Content-Type: application/json' -d '{body}'")
|
|
try:
|
|
data = json.loads(out)
|
|
token = data.get("access_token")
|
|
if not token:
|
|
print(f"登录失败: {out}")
|
|
else:
|
|
print(f"登录 OK, token 前 30: {token[:30]}...")
|
|
# 拉 articles
|
|
out2 = run(f"curl -s -H 'Authorization: Bearer {token}' 'http://localhost/api/v1/articles?limit=3'")
|
|
ad = json.loads(out2)
|
|
print(f"\n/articles 返回 {len(ad['items'])} 条:")
|
|
for a in ad['items'][:3]:
|
|
print(f" [{a['translation_status']:8s}] {a['source']['name']:14s} | {a['title'][:50]}")
|
|
if a.get('title_zh'):
|
|
print(f" zh: {a['title_zh'][:50]}")
|
|
# /me
|
|
me = json.loads(run(f"curl -s -H 'Authorization: Bearer {token}' 'http://localhost/api/v1/me'"))
|
|
print(f"\n/me: {me}")
|
|
# /me/usage
|
|
u = json.loads(run(f"curl -s -H 'Authorization: Bearer {token}' 'http://localhost/api/v1/me/usage'"))
|
|
print(f"/me/usage: {u}")
|
|
except Exception as e:
|
|
print(f"parse err: {e}\n raw: {out}")
|
|
c.close()
|