chore: refresh public mirror
This commit is contained in:
chen qiang
+150
-1
@@ -20,6 +20,11 @@ import io
|
||||
import re
|
||||
import html
|
||||
import requests
|
||||
import base64
|
||||
import hashlib
|
||||
import platform
|
||||
import uuid
|
||||
import secrets
|
||||
from typing import Optional, Dict, Any, List, Tuple
|
||||
|
||||
import argparse
|
||||
@@ -1996,6 +2001,103 @@ def cmd_login(secret_key: str, client_code: str = ""):
|
||||
print(f" 可切换 client: {', '.join(c['code'] for c in client_list)}")
|
||||
|
||||
|
||||
def _device_key_path() -> str:
|
||||
return os.path.join(get_script_dir(), "device-key.json")
|
||||
|
||||
|
||||
def _load_device_key() -> Dict[str, Any]:
|
||||
with open(_device_key_path(), "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def _save_device_key(item: Dict[str, Any]) -> None:
|
||||
path = _device_key_path()
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(item, f, ensure_ascii=False, indent=2)
|
||||
try:
|
||||
os.chmod(path, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _device_identity() -> Tuple[str, str]:
|
||||
mac = ":".join(f"{(uuid.getnode() >> shift) & 0xff:02X}" for shift in range(40, -1, -8))
|
||||
return hashlib.sha256(mac.encode("ascii")).hexdigest(), f"{mac[:8]}****{mac[-5:]}"
|
||||
|
||||
|
||||
def cmd_device_register(device_name: str = ""):
|
||||
"""Generate an Ed25519 key locally and register this terminal as PENDING."""
|
||||
try:
|
||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||||
except ImportError:
|
||||
print("❌ 需要安装 cryptography: python -m pip install cryptography")
|
||||
return
|
||||
cfg = get_config() or {}
|
||||
if not ensure_logged_in(cfg):
|
||||
print("❌ 当前登录态已失效，请先执行 login")
|
||||
return
|
||||
try:
|
||||
key = _load_device_key()
|
||||
private = Ed25519PrivateKey.from_private_bytes(base64.b64decode(key["private_key"]))
|
||||
except (OSError, KeyError, ValueError, TypeError):
|
||||
private = Ed25519PrivateKey.generate()
|
||||
public = private.public_key().public_bytes_raw()
|
||||
key = {"device_id": "device-" + secrets.token_hex(12), "private_key": base64.b64encode(private.private_bytes_raw()).decode("ascii"), "public_key": base64.b64encode(public).decode("ascii"), "key_fingerprint": hashlib.sha256(public).hexdigest()}
|
||||
_save_device_key(key)
|
||||
mac_hash, mac_masked = _device_identity()
|
||||
body = {"device_id": key["device_id"], "device_name": device_name or platform.node() or "Trusted device", "computer_name": platform.node(), "os_name": platform.platform(), "key_algorithm": "ED25519", "public_key": key["public_key"], "key_fingerprint": key["key_fingerprint"], "mac_hash": mac_hash, "mac_masked": mac_masked, "description": "Registered by oracle-jump-query skill"}
|
||||
try:
|
||||
resp = requests.post(f"{cfg.get('transit_url', TRANSIT_URL)}/api/device/register", json=body, headers=make_headers(), timeout=20)
|
||||
data = resp.json()
|
||||
except (requests.RequestException, ValueError) as exc:
|
||||
print(f"❌ 可信设备注册失败: {exc}")
|
||||
return
|
||||
if not data.get("success"):
|
||||
print(f"❌ 可信设备注册失败: {data.get('error', resp.text)}")
|
||||
return
|
||||
print("✅ 可信设备注册请求已提交，当前状态为 PENDING")
|
||||
print(f" device_id: {key['device_id']}")
|
||||
print(f" fingerprint: {key['key_fingerprint']}")
|
||||
print(" 请在后台表单审批为 APPROVED 后执行 device_login")
|
||||
|
||||
|
||||
def cmd_device_login(client_code: str = ""):
|
||||
"""Use the locally stored Ed25519 key to perform trusted-device login."""
|
||||
try:
|
||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||||
key = _load_device_key()
|
||||
private = Ed25519PrivateKey.from_private_bytes(base64.b64decode(key["private_key"]))
|
||||
except (ImportError, OSError, KeyError, ValueError, TypeError) as exc:
|
||||
print(f"❌ 本机可信设备密钥不可用: {exc}")
|
||||
return
|
||||
transit_url = (get_config() or {}).get("transit_url", TRANSIT_URL)
|
||||
identity = {"device_id": key["device_id"], "key_fingerprint": key["key_fingerprint"]}
|
||||
try:
|
||||
challenge_resp = requests.post(f"{transit_url}/api/device/challenge", json=identity, timeout=20)
|
||||
challenge_data = challenge_resp.json()
|
||||
if not challenge_data.get("success"):
|
||||
print(f"❌ 获取可信设备挑战失败: {challenge_data.get('error', challenge_resp.text)}")
|
||||
return
|
||||
challenge = challenge_data["challenge"]
|
||||
signature = base64.b64encode(private.sign(challenge.encode("utf-8"))).decode("ascii")
|
||||
resp = requests.post(f"{transit_url}/api/device/login", json={**identity, "challenge": challenge, "signature": signature, "client_code": client_code}, timeout=20)
|
||||
data = resp.json()
|
||||
except (requests.RequestException, ValueError) as exc:
|
||||
print(f"❌ 可信设备登录失败: {exc}")
|
||||
return
|
||||
if not data.get("success"):
|
||||
print(f"❌ 可信设备登录失败: {data.get('error', resp.text)}")
|
||||
return
|
||||
cfg = get_config() or {}
|
||||
cfg.update({"access_token": data.get("access_token", ""), "expires_at": data.get("expires_at", ""), "user_name": data.get("user", {}).get("name", ""), "client_code": data.get("current_client", {}).get("code", client_code), "client_title": data.get("current_client", {}).get("title", "")})
|
||||
cfg["server_id"] = cfg.get("client_code", "")
|
||||
cfg["client_list"] = data.get("client_list", [])
|
||||
save_config(cfg)
|
||||
print("✅ 可信设备登录成功")
|
||||
print(f" 用户: {cfg.get('user_name', '')}")
|
||||
print(f" 指向 client: {cfg.get('client_code', '')}")
|
||||
|
||||
|
||||
def cmd_logout():
|
||||
"""登出，清除登录信息"""
|
||||
cfg = get_config() or {}
|
||||
@@ -2144,8 +2246,37 @@ def is_token_expired(cfg: dict) -> bool:
|
||||
return now >= expires_at
|
||||
|
||||
|
||||
def _auto_device_login(cfg: Dict[str, Any]) -> bool:
|
||||
"""Refresh the transit session with the locally approved trusted device."""
|
||||
try:
|
||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||||
key = _load_device_key()
|
||||
private = Ed25519PrivateKey.from_private_bytes(base64.b64decode(key["private_key"]))
|
||||
identity = {"device_id": key["device_id"], "key_fingerprint": key["key_fingerprint"]}
|
||||
transit_url = cfg.get("transit_url", TRANSIT_URL)
|
||||
challenge_data = requests.post(f"{transit_url}/api/device/challenge", json=identity, timeout=20).json()
|
||||
if not challenge_data.get("success"):
|
||||
return False
|
||||
challenge = challenge_data["challenge"]
|
||||
signature = base64.b64encode(private.sign(challenge.encode("utf-8"))).decode("ascii")
|
||||
data = requests.post(f"{transit_url}/api/device/login", json={**identity, "challenge": challenge, "signature": signature, "client_code": cfg.get("client_code", "")}, timeout=20).json()
|
||||
if not data.get("success") or not data.get("access_token"):
|
||||
return False
|
||||
current = data.get("current_client", {})
|
||||
cfg.update({"access_token": data["access_token"], "expires_at": data.get("expires_at", ""), "user_name": data.get("user", {}).get("name", cfg.get("user_name", "")), "client_code": current.get("code", cfg.get("client_code", "")), "client_title": current.get("title", cfg.get("client_title", "")), "client_list": data.get("client_list", cfg.get("client_list", []))})
|
||||
cfg["server_id"] = cfg.get("client_code", cfg.get("server_id", ""))
|
||||
save_config(cfg)
|
||||
print("✅ 中转 token 已通过可信设备自动续取")
|
||||
return True
|
||||
except (ImportError, OSError, KeyError, ValueError, TypeError, requests.RequestException):
|
||||
return False
|
||||
|
||||
|
||||
def ensure_logged_in(cfg: Optional[dict] = None) -> bool:
|
||||
cfg = cfg or get_config() or {}
|
||||
if not cfg.get("access_token") or is_token_expired(cfg):
|
||||
if _auto_device_login(cfg):
|
||||
return True
|
||||
if not cfg.get("access_token"):
|
||||
print("❌ 未登录，请先执行: python oracle_skill.py login <secretKey> [clientCode]")
|
||||
return False
|
||||
@@ -3191,6 +3322,18 @@ def cmd_capabilities(args):
|
||||
"required": []
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "device_register",
|
||||
"description": "Generate a local Ed25519 device key and submit a trusted-device registration request.",
|
||||
"parameters": {"type": "object", "properties": {"deviceName": {"type": "string", "description": "Optional device display name."}}},
|
||||
"required": []
|
||||
},
|
||||
{
|
||||
"name": "device_login",
|
||||
"description": "Authenticate through an approved trusted device using a local Ed25519 private key.",
|
||||
"parameters": {"type": "object", "properties": {"clientCode": {"type": "string", "description": "Optional client/server code."}}},
|
||||
"required": []
|
||||
},
|
||||
{
|
||||
"name": "awr_status",
|
||||
"description": "Read AWR authorization, readiness and latest generated report status without generating a report.",
|
||||
@@ -3271,7 +3414,7 @@ def cmd_capabilities(args):
|
||||
"analyze", "list", "source", "describe",
|
||||
"search", "query", "discover", "nl2sql",
|
||||
"perm", "qperm", "login", "logout",
|
||||
"status", "switch", "clients", "ops", "ops_report", "inspection_report", "inspection_latest", "inspection_get", "awr_status", "awr_list", "awr_download", "agent_update", "capabilities"
|
||||
"status", "switch", "clients", "device_register", "device_login", "ops", "ops_report", "inspection_report", "inspection_latest", "inspection_get", "awr_status", "awr_list", "awr_download", "agent_update", "capabilities"
|
||||
],
|
||||
"python_version": "3.6+",
|
||||
"dependencies": ["requests"]
|
||||
@@ -3410,6 +3553,12 @@ def main():
|
||||
|
||||
elif cmd == "logout":
|
||||
cmd_logout()
|
||||
|
||||
elif cmd == "device_register":
|
||||
cmd_device_register(" ".join(sys.argv[2:]) if len(sys.argv) >= 3 else "")
|
||||
|
||||
elif cmd == "device_login":
|
||||
cmd_device_login(sys.argv[2] if len(sys.argv) >= 3 else "")
|
||||
|
||||
elif cmd == "status":
|
||||
cmd_status()
|
||||
|
||||
Reference in New Issue
Block a user