mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-30 11:49:48 +02:00
1880 lines
78 KiB
Python
1880 lines
78 KiB
Python
# coding: utf-8
|
|
"""Panel operation toolset - 44 tools registered to ToolRegistry"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import subprocess
|
|
import time
|
|
from typing import List
|
|
|
|
from . import register_tool
|
|
from .base import _xml_response
|
|
|
|
_panel_path = '/www/server/panel'
|
|
os.chdir(_panel_path)
|
|
sys.path.insert(0, _panel_path)
|
|
sys.path.insert(0, _panel_path + "/class/")
|
|
sys.path.insert(0, _panel_path + "/class_v2/")
|
|
|
|
import public
|
|
|
|
try:
|
|
from public.hook_import import hook_import
|
|
|
|
hook_import()
|
|
except:
|
|
pass
|
|
|
|
from public import lang
|
|
|
|
|
|
def _to_obj(params: dict):
|
|
"""Convert dict to panel obj object"""
|
|
return public.to_dict_obj(params)
|
|
|
|
|
|
class _ServiceStatusEnricher:
|
|
"""Backfill s_status for services not covered by get_soft_status.
|
|
|
|
Layered on top of the get_soft_status result, reusing panelPlugin detectors
|
|
for the services it misses (php/memcached/tomcat/phpmyadmin + a generic
|
|
fallback). Output stays structurally identical — same outer keys and the
|
|
same inner 9 keys; on any error the wrapper is returned unchanged so the
|
|
tool never breaks the base output.
|
|
"""
|
|
|
|
_COVERED = frozenset({
|
|
'nginx', 'mysql', 'apache', 'pure-ftpd', 'redis',
|
|
'pgsql', 'openlitespeed', 'mongodb',
|
|
})
|
|
_PHP_RE = re.compile(r'^php(?:-fpm)?(?:-?(\d{1,2})(?:\.\d+)?)?$')
|
|
_TITLES = {'tomcat': 'Tomcat', 'phpmyadmin': 'phpMyAdmin'}
|
|
|
|
def __init__(self, plugin):
|
|
self.plugin = plugin
|
|
|
|
@classmethod
|
|
def enrich(cls, wrapper, name):
|
|
if not isinstance(wrapper, dict):
|
|
return wrapper
|
|
inner = wrapper.get('message')
|
|
if not isinstance(inner, dict):
|
|
return wrapper
|
|
raw = (name or '').strip()
|
|
family, phpver = cls._classify(raw)
|
|
if not family or family in cls._COVERED:
|
|
return wrapper
|
|
try:
|
|
from class_v2.panel_plugin_v2 import panelPlugin
|
|
enricher = cls(panelPlugin())
|
|
except Exception:
|
|
return wrapper
|
|
|
|
if family == 'php':
|
|
# get_soft_status judges install via /www/server/<name>, which misses
|
|
# php-fpm (real dir is /www/server/php/<ver>); re-judge via dir scan.
|
|
running, version, installed = enricher._php(phpver)
|
|
if installed:
|
|
inner['status'] = True
|
|
inner['setup'] = True
|
|
inner['s_status'] = bool(running)
|
|
if version:
|
|
inner['version'] = version
|
|
if inner.get('title', '') in ('', raw):
|
|
inner['title'] = 'PHP' + (phpver or '')
|
|
return wrapper
|
|
|
|
# Non-php: leave uninstalled services untouched (s_status stays False).
|
|
if not inner.get('status'):
|
|
return wrapper
|
|
try:
|
|
running, version = enricher._detect(family, raw, phpver)
|
|
except Exception:
|
|
return wrapper
|
|
if family in ('memcached', 'tomcat', 'phpmyadmin'):
|
|
inner['s_status'] = bool(running)
|
|
if version and not inner.get('version'):
|
|
inner['version'] = version
|
|
if inner.get('title', '') in ('', raw):
|
|
inner['title'] = cls._TITLES.get(family, inner.get('title', ''))
|
|
elif running:
|
|
# Generic fallback: only flip to True when running is detected.
|
|
inner['s_status'] = True
|
|
return wrapper
|
|
|
|
@classmethod
|
|
def query(cls, names, fetch=None):
|
|
"""High-level entry: return a newline-joined status report, one line per
|
|
service (e.g. 'nginx: installed, running, v1.24'). Encapsulates
|
|
get_soft_status + enrich + describe."""
|
|
if fetch is None:
|
|
from class_v2.panelModelV2.publicModel import main as PublicMain
|
|
fetch = PublicMain().get_soft_status
|
|
lines = []
|
|
for svc in names:
|
|
data = cls.enrich(fetch(_to_obj({"name": svc})), svc)
|
|
inner = data.get('message', {}) if isinstance(data, dict) else {}
|
|
lines.append(cls._describe(svc, inner))
|
|
return '\n'.join(lines)
|
|
|
|
@classmethod
|
|
def _describe(cls, svc, inner):
|
|
"""Render one service's status as a model-friendly line."""
|
|
if not isinstance(inner, dict) or not inner.get('status'):
|
|
return '{}: not installed'.format(svc)
|
|
parts = ['installed', 'running' if inner.get('s_status') else 'stopped']
|
|
version = inner.get('version', '')
|
|
if version:
|
|
parts.append('v' + version)
|
|
return '{}: {}'.format(svc, ', '.join(parts))
|
|
|
|
@classmethod
|
|
def _classify(cls, raw):
|
|
"""Return (family, php_version). 'php' for any php/php-fpm/php-<ver>
|
|
variant, else the raw name; php_version is dot-stripped (e.g. '74')."""
|
|
n = (raw or '').strip().lower()
|
|
if not n:
|
|
return ('', None)
|
|
m = cls._PHP_RE.match(n)
|
|
if m:
|
|
return ('php', m.group(1))
|
|
return (n, None)
|
|
|
|
def _detect(self, family, raw, phpver):
|
|
"""Return (running, version) for memcached/tomcat/phpmyadmin/generic."""
|
|
if family == 'memcached':
|
|
return (self._exists('memcached'), '')
|
|
if family == 'tomcat':
|
|
return (self._exists('jsvc') or self._exists('java'), '')
|
|
if family == 'phpmyadmin':
|
|
try:
|
|
return (bool(self.plugin.get_phpmyadmin_stat()), '')
|
|
except Exception:
|
|
return (False, '')
|
|
# Generic fallback: the raw name, its separator-stripped form, or a
|
|
# common daemon alias.
|
|
candidates = [raw, re.sub(r'[-_.]', '', raw),
|
|
{'apache': 'httpd', 'mysql': 'mysqld'}.get(raw, '')]
|
|
return (any(self._exists(c) for c in candidates if c), '')
|
|
|
|
def _exists(self, pname):
|
|
try:
|
|
return bool(self.plugin.process_exists(pname))
|
|
except Exception:
|
|
return False
|
|
|
|
def _php(self, ver):
|
|
"""Detect php-fpm via installed /www/server/php/<digits> dirs.
|
|
|
|
When ver is given only that version is checked; version joins detected
|
|
versions like '7.4/8.0'."""
|
|
base = '/www/server/php'
|
|
installed = []
|
|
if os.path.isdir(base):
|
|
installed = [d for d in os.listdir(base)
|
|
if d.isdigit() and os.path.isdir(os.path.join(base, d))]
|
|
if ver:
|
|
installed = [v for v in installed if v == ver]
|
|
if not installed:
|
|
return (False, '', False)
|
|
running, versions = False, []
|
|
for v in installed:
|
|
try:
|
|
if self.plugin.get_php_status(v):
|
|
running = True
|
|
except Exception:
|
|
continue
|
|
try:
|
|
with open(os.path.join(base, v, 'version.pl'), 'r') as f:
|
|
content = f.read().strip()
|
|
if content:
|
|
versions.append(content)
|
|
except Exception:
|
|
pass
|
|
return (running, '/'.join(versions), True)
|
|
|
|
|
|
def _truncate_result(data, limit):
|
|
"""Best-effort truncate a tool result to `limit` items without losing info.
|
|
Handles plain lists and common wrappers ({data/rows/list/records: [...]}).
|
|
Returns data unchanged when no list is found (info-preserving fallback)."""
|
|
def _slice(lst):
|
|
total = len(lst)
|
|
if total > limit:
|
|
lst = lst[:limit]
|
|
lst.append({"_truncated": True, "total": total, "limit": limit})
|
|
return lst
|
|
if isinstance(data, list):
|
|
return _slice(data)
|
|
if isinstance(data, dict):
|
|
for key in ("data", "rows", "list", "records"):
|
|
v = data.get(key)
|
|
if isinstance(v, list):
|
|
data[key] = _slice(v)
|
|
return data
|
|
return data
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# System Diagnostics
|
|
# ═══════════════════════════════════════════════════════════════
|
|
@register_tool(category="System", name_cn="Get System Resources", risk_level="low")
|
|
def GetSystemResources() -> str:
|
|
"""
|
|
Get real-time system resource metrics: CPU load, memory, disk usage, OS version.
|
|
|
|
Returns: Load Avg (1m/5m/15m), Memory (used/total + %), Disk / (used/total + %), OS version.
|
|
|
|
Workflow: GetSystemResources -> GetTopProcesses (for process-level diagnosis).
|
|
"""
|
|
try:
|
|
# Load Average
|
|
try:
|
|
load1, load5, load15 = os.getloadavg()
|
|
load_info = f"Load Avg: {load1:.2f}, {load5:.2f}, {load15:.2f}"
|
|
except OSError:
|
|
load_info = "Load Avg: N/A (Windows?)"
|
|
|
|
# Memory
|
|
mem_info = "Mem: Unknown"
|
|
if os.path.exists('/proc/meminfo'):
|
|
with open('/proc/meminfo', 'r') as f:
|
|
lines = f.readlines()
|
|
total = 0
|
|
available = 0
|
|
for line in lines:
|
|
if 'MemTotal' in line:
|
|
total = int(line.split()[1]) // 1024 # MB
|
|
if 'MemAvailable' in line:
|
|
available = int(line.split()[1]) // 1024 # MB
|
|
used = total - available
|
|
percent = (used / total * 100) if total > 0 else 0
|
|
mem_info = f"Mem: {used}MB/{total}MB ({percent:.1f}%)"
|
|
|
|
# Disk
|
|
import shutil
|
|
disk = shutil.disk_usage("/")
|
|
total_gb = disk.total // (1024 ** 3)
|
|
used_gb = disk.used // (1024 ** 3)
|
|
disk_percent = (disk.used / disk.total * 100)
|
|
disk_info = f"Disk (/): {used_gb}GB/{total_gb}GB ({disk_percent:.1f}%)"
|
|
os_info = public.get_os_version()
|
|
|
|
result = f"{load_info}\n{mem_info}\n{disk_info}\nOS: {os_info}"
|
|
return _xml_response("GetSystemResources", "done", result)
|
|
except Exception as e:
|
|
return _xml_response("GetSystemResources", "error", f"Error getting resources: {str(e)}")
|
|
|
|
|
|
@register_tool(category="System", name_cn="Get Public IP", risk_level="low")
|
|
def GetPublicIP() -> str:
|
|
"""
|
|
Get the server's public IP (via external probe) and local network IP.
|
|
|
|
Returns both addresses for comparison:
|
|
- Public IP: the address the internet sees, used for DNS A-record, SSL application, firewall rules.
|
|
- Local IP: the server's primary network interface address (may differ from public IP behind NAT).
|
|
|
|
If public IP detection fails, only local IP is returned.
|
|
"""
|
|
result_parts = []
|
|
|
|
# 公网 IP
|
|
public_ip = None
|
|
for api_url in ['https://api.ipify.org', 'https://ifconfig.me', 'https://ip.sb']:
|
|
try:
|
|
p = subprocess.run(
|
|
['curl', '-s', '--connect-timeout', '3', api_url],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
candidate = p.stdout.strip()
|
|
if candidate and re.match(r'^\d{1,3}(\.\d{1,3}){3}$', candidate):
|
|
public_ip = candidate
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
if public_ip:
|
|
result_parts.append(f"Public IP: {public_ip}")
|
|
else:
|
|
result_parts.append("Public IP: N/A (detection failed)")
|
|
|
|
# 本机网卡IP
|
|
try:
|
|
local_ip = public.GetLocalIp()
|
|
result_parts.append(f"Local IP: {local_ip}")
|
|
except Exception:
|
|
result_parts.append("Local IP: N/A")
|
|
|
|
return _xml_response("GetPublicIP", "done", "\n".join(result_parts))
|
|
|
|
|
|
@register_tool(category="System", name_cn="Get Top 10 Processes", risk_level="low")
|
|
def GetTopProcesses() -> str:
|
|
"""
|
|
Get top 10 processes by CPU and top 10 by memory.
|
|
|
|
Returns: Two sections - CPU TOP 10 and Memory TOP 10 (PID, user, %cpu, %mem, command).
|
|
|
|
Workflow: GetSystemResources -> GetTopProcesses.
|
|
"""
|
|
|
|
def _run_shell_cmd(command: list, timeout: int = 300) -> tuple:
|
|
"""
|
|
Common function to execute shell commands.
|
|
Returns (success: bool, output: str)
|
|
"""
|
|
try:
|
|
# Use shell=False for security when passing a list
|
|
result = subprocess.run(
|
|
command,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout
|
|
)
|
|
|
|
output = result.stdout.strip()
|
|
if not output:
|
|
output = result.stderr.strip()
|
|
|
|
return result.returncode == 0, output
|
|
except subprocess.TimeoutExpired:
|
|
return False, f"Error: Command timed out after {timeout} seconds."
|
|
except FileNotFoundError:
|
|
return False, f"Error: Command not found: {command[0]}"
|
|
except Exception as e:
|
|
return False, f"Error executing command: {str(e)}"
|
|
|
|
output_parts = []
|
|
|
|
# 1. CPU Top 10
|
|
success_cpu, output_cpu = _run_shell_cmd(["ps", "-eo", "pid,user,%cpu,%mem,command", "--sort=-%cpu"])
|
|
if success_cpu:
|
|
lines = output_cpu.strip().splitlines()
|
|
header = lines[0] if lines else ""
|
|
top10 = lines[1:11]
|
|
output_parts.append("--- CPU Usage TOP 10 ---")
|
|
output_parts.append(header)
|
|
output_parts.extend(top10)
|
|
else:
|
|
output_parts.append(f"{lang('Failed to get CPU TOP 10')}: {output_cpu}")
|
|
|
|
output_parts.append("") # 空行分隔
|
|
|
|
# 2. Memory Top 10
|
|
success_mem, output_mem = _run_shell_cmd(["ps", "-eo", "pid,user,%cpu,%mem,command", "--sort=-%mem"])
|
|
if success_mem:
|
|
lines = output_mem.strip().splitlines()
|
|
header = lines[0] if lines else ""
|
|
top10 = lines[1:11]
|
|
output_parts.append("--- Memory Usage TOP 10 ---")
|
|
output_parts.append(header)
|
|
output_parts.extend(top10)
|
|
else:
|
|
output_parts.append(f"{lang('Failed to get Memory TOP 10')}: {output_mem}")
|
|
|
|
return _xml_response("GetTopProcesses", "done", "\n".join(output_parts))
|
|
|
|
|
|
# Website Management
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="Website", name_cn="List Websites", risk_level="low")
|
|
def SiteList(search: str = "", limit: int = 100, include_health: bool = False,
|
|
site_id_list: List[int] = None) -> str:
|
|
"""
|
|
List all websites. Always call this first before any website operation.
|
|
|
|
Returns: JSON array with id, name (domain), project_type, path, status, ps.
|
|
Set include_health=True to also return ssl/php_version/backup_count/domain_count/domains per site.
|
|
|
|
Workflow: SiteList -> SiteDelete / WebSiteBackup / AutoSSL / GetSitesConf / GetSitesLogs.
|
|
ALL downstream website tools take `id` from this list as the primary key.
|
|
|
|
Args:
|
|
search: Optional domain filter keyword. e.g. "example.com".
|
|
limit: Max items to return (default 100). Items beyond are omitted.
|
|
include_health: Also return ssl/php_version/backup_count/domain_count/domains per site. Default False.
|
|
site_id_list: Optional list of specific site IDs.
|
|
"""
|
|
try:
|
|
_fields = 'id,name,project_type,path,status,ps'
|
|
if site_id_list:
|
|
ids_in = [int(i) for i in site_id_list]
|
|
ph = ','.join(['?'] * len(ids_in))
|
|
sites = public.M('sites').field(_fields).where('id in ({})'.format(ph), tuple(ids_in)).select()
|
|
elif search:
|
|
sites = public.M('sites').field(_fields).where('name like ?', '%{}%'.format(search)).select()
|
|
else:
|
|
sites = public.M('sites').field(_fields).select()
|
|
|
|
# 轻量索引模式: 直出
|
|
if not include_health:
|
|
sites = _truncate_result(sites, limit)
|
|
return _xml_response("SiteList", "done", json.dumps(sites, ensure_ascii=False))
|
|
|
|
# 诊断模式: 批量查 + 逐站富化 (ssl/php_version/backup_count/domain_count/domains)
|
|
if not sites:
|
|
return _xml_response("SiteList", "done", json.dumps([], ensure_ascii=False))
|
|
|
|
from class_v2.data_v2 import data as _Data
|
|
|
|
ids = [s['id'] for s in sites]
|
|
|
|
bak_map = {}
|
|
try:
|
|
bak_rows = public.S('backup').where_in('pid', ids).where("type", "0").group('pid').field(
|
|
'pid', 'count(*) as cnt', 'max(addtime) as last_backup_time').select()
|
|
bak_map = {b['pid']: b for b in bak_rows}
|
|
except Exception:
|
|
pass
|
|
|
|
domain_map = {}
|
|
try:
|
|
dom_rows = public.S('domain').where_in('pid', ids).field('pid', 'name').select()
|
|
for _d in dom_rows:
|
|
domain_map.setdefault(_d['pid'], []).append(_d['name'])
|
|
except Exception:
|
|
pass
|
|
|
|
d = _Data()
|
|
|
|
def _enrich(site):
|
|
try:
|
|
_ssl = d.get_site_ssl_info(site['name'])
|
|
# 只保留诊断过期所需字段, 砍掉 issuer/notBefore/dns/subject
|
|
site['ssl'] = {'endtime': _ssl['endtime'], 'notAfter': _ssl['notAfter']} \
|
|
if isinstance(_ssl, dict) else _ssl
|
|
except Exception:
|
|
site['ssl'] = -1
|
|
try:
|
|
site['php_version'] = d.get_php_version(site['name'])
|
|
except Exception:
|
|
site['php_version'] = 'Static'
|
|
bak = bak_map.get(site['id'], {})
|
|
site['backup_count'] = bak.get('cnt', 0)
|
|
site['last_backup_time'] = bak.get('last_backup_time') or ''
|
|
_doms = domain_map.get(site['id'], [])
|
|
site['domains'] = _doms
|
|
site['domain_count'] = len(_doms)
|
|
return site
|
|
|
|
sites = [_enrich(s) for s in sites]
|
|
|
|
sites = _truncate_result(sites, limit)
|
|
return _xml_response("SiteList", "done", json.dumps(sites, ensure_ascii=False, default=str))
|
|
except Exception as e:
|
|
return _xml_response("SiteList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Website", name_cn="Get PHP Versions", risk_level="low")
|
|
def SiteGetPHPVersions() -> str:
|
|
"""
|
|
Get PHP versions installed on the server.
|
|
|
|
Returns: JSON array of version codes. e.g. ["00","73","74","80","81","82","83","84"].
|
|
"00" = pure static, "other" = custom.
|
|
|
|
Workflow: SiteGetPHPVersions -> SiteCreate (pass version code).
|
|
"""
|
|
try:
|
|
import panel_site_v2
|
|
data = panel_site_v2.panelSite().GetPHPVersion(public.to_dict_obj({}))
|
|
return _xml_response("SiteGetPHPVersions", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("SiteGetPHPVersions", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Website", name_cn="Create Website", risk_level="medium")
|
|
def SiteCreate(domain: str, port: int = 80, version: str = "", ps: str = "", path: str = "",
|
|
ftp: bool = False, ftp_username: str = "", ftp_password: str = "",
|
|
sql: str = "", datauser: str = "", datapassword: str = "",
|
|
codeing: str = "utf8", type_id: int = 0,
|
|
force_ssl: int = 0,
|
|
is_create_default_file: bool = True) -> str:
|
|
"""
|
|
Create a new website with domain binding, optional FTP and database. Backend auto-applies SSL asynchronously (ssl_auto=1); not the AutoSSL tool.
|
|
|
|
Args:
|
|
domain: Primary domain. Required. e.g. "example.com" or "blog.example.com".
|
|
port: Listen port. Default 80.
|
|
version: PHP version code. Default "" (pure static). Call SiteGetPHPVersions first.
|
|
ps: Remark. Optional.
|
|
path: Website root directory. Optional, auto-generated.
|
|
ftp: Create FTP account. Default False.
|
|
ftp_username: FTP username. Optional when ftp=True, auto-generated if omitted.
|
|
ftp_password: FTP password. Optional when ftp=True, auto-generated if omitted.
|
|
sql: Database type. "MySQL" to create DB, "" to skip. Default "".
|
|
datauser: Database username. Optional when sql="MySQL", auto-generated if omitted.
|
|
datapassword: Database password. Optional when sql="MySQL", auto-generated if omitted.
|
|
codeing: Database encoding. Default "utf8".
|
|
type_id: Site type ID. Default 0.
|
|
force_ssl: Force HTTPS redirect. Default 0.
|
|
is_create_default_file: Create default index file. Default True.
|
|
|
|
Workflow: SiteGetPHPVersions -> SiteCreate.
|
|
"""
|
|
try:
|
|
import panel_site_v2
|
|
# 构造 webname JSON (后端要求 JSON 字符串)
|
|
webname_obj = {
|
|
"domain": domain,
|
|
"domainlist": [],
|
|
"count": 0,
|
|
}
|
|
params = {
|
|
"webname": json.dumps(webname_obj, ensure_ascii=False),
|
|
"port": str(port),
|
|
"type": "PHP",
|
|
"version": version or "00",
|
|
"ps": ps or domain.replace(".", "_"),
|
|
"path": path or f"/www/wwwroot/{domain}",
|
|
"ftp": ftp,
|
|
"sql": sql or "",
|
|
"codeing": codeing,
|
|
"type_id": type_id,
|
|
"force_ssl": force_ssl,
|
|
"ssl_auto": "1",
|
|
"is_create_default_file": is_create_default_file,
|
|
}
|
|
if ftp:
|
|
params["ftp_username"] = ftp_username or f"ftp_{domain.replace('.', '_')}"
|
|
params["ftp_password"] = ftp_password or __import__('secrets').token_hex(8)
|
|
if sql:
|
|
params["datauser"] = datauser or f"sql_{domain.replace('.', '_')}"
|
|
params["datapassword"] = datapassword or __import__('secrets').token_hex(8)
|
|
data = panel_site_v2.panelSite().AddSite(_to_obj(params))
|
|
return _xml_response("SiteCreate", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("SiteCreate", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Website", name_cn="Delete Website", risk_level="high")
|
|
def SiteDelete(id: int, webname: str = "") -> str:
|
|
"""
|
|
Delete a website. Irreversible.
|
|
|
|
Args:
|
|
id: Website ID from SiteList. Required.
|
|
webname: Website domain name. Optional.
|
|
|
|
Workflow: SiteList -> SiteDelete.
|
|
"""
|
|
try:
|
|
import panel_site_v2
|
|
params = {"id": id}
|
|
if webname:
|
|
params["webname"] = webname
|
|
data = panel_site_v2.panelSite().DeleteSite(_to_obj(params))
|
|
return _xml_response("SiteDelete", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("SiteDelete", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Website", name_cn="Set Website PHP Version", risk_level="high")
|
|
def SiteSetPHPVersion(site_id: int, version: str) -> str:
|
|
"""
|
|
Switch a website's PHP version. May briefly interrupt the site while the web server reloads.
|
|
|
|
Args:
|
|
site_id: Site ID from SiteList. Required.
|
|
version: PHP version code from SiteGetPHPVersions. Required. e.g. "74","83"; "00"=pure static, "other"=custom.
|
|
|
|
Workflow: SiteGetPHPVersions -> SiteSetPHPVersion.
|
|
|
|
Prefer this over manually editing the vhost PHP include.
|
|
"""
|
|
try:
|
|
site = public.M('sites').where('id=?', site_id).field('name').find()
|
|
if not site:
|
|
return _xml_response("SiteSetPHPVersion", "error", "Site {} not found".format(site_id))
|
|
import panel_site_v2
|
|
data = panel_site_v2.panelSite().SetPHPVersion(
|
|
_to_obj({"siteName": site["name"], "version": version}))
|
|
return _xml_response("SiteSetPHPVersion", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("SiteSetPHPVersion", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Website", name_cn="Get Website Config", risk_level="low")
|
|
def GetSitesConf(site_id_list: List[int] = None) -> str:
|
|
"""
|
|
Get nginx/Apache vhost configuration for one or more websites.
|
|
|
|
Use for: diagnosing 403/502/500 errors, checking rewrite rules, SSL, reverse proxy config.
|
|
|
|
Args:
|
|
site_id_list: List of site IDs from SiteList. Pass SPECIFIC sites; Empty = ALL sites
|
|
(returns every vhost config and can be very large - prefer targeted queries).
|
|
|
|
Workflow: SiteList -> GetSitesConf (takes id; resolves domain internally).
|
|
"""
|
|
import json
|
|
if not site_id_list:
|
|
all_sites = public.M('sites').field('id,name,project_type').select()
|
|
if not all_sites:
|
|
return _xml_response("GetSitesConf", "done",
|
|
json.dumps({"error": "No sites found in panel."}, ensure_ascii=False, indent=2))
|
|
site_id_list = [s['id'] for s in all_sites]
|
|
|
|
results = {}
|
|
last_site_name = None
|
|
for sid in site_id_list:
|
|
site_data = public.M('sites').field('name,project_type').where("id=?", int(sid)).find()
|
|
if not site_data:
|
|
results[str(sid)] = {"error": "Site not found in panel."}
|
|
continue
|
|
|
|
site_name = site_data['name']
|
|
last_site_name = site_name
|
|
project_type = site_data['project_type'].lower()
|
|
prefix = '' if project_type in ('php', 'proxy', 'phpmod', 'wp2') else project_type + '_'
|
|
|
|
conf_path = f"/www/server/panel/vhost/nginx/{prefix}{site_name}.conf"
|
|
if not os.path.exists(conf_path):
|
|
conf_path = f"/www/server/panel/vhost/apache/{prefix}{site_name}.conf"
|
|
if not os.path.exists(conf_path):
|
|
results[site_name] = {"error": "Configuration file not found."}
|
|
continue
|
|
|
|
with open(conf_path, 'r') as f:
|
|
results[site_name] = f.read()
|
|
|
|
if len(site_id_list) == 1 and last_site_name in results:
|
|
val = results[last_site_name]
|
|
if not (isinstance(val, dict) and "error" in val):
|
|
return _xml_response("GetSitesConf", "done", val)
|
|
return _xml_response("GetSitesConf", "done", json.dumps(results, ensure_ascii=False, indent=2))
|
|
|
|
|
|
@register_tool(category="Website", name_cn="Get Website Access Logs", risk_level="low")
|
|
def GetSitesLogs(site_id_list: List[int] = None) -> str:
|
|
"""
|
|
Get recent access logs for websites.
|
|
|
|
Use for: analyzing traffic, HTTP status codes, diagnosing 404/500, detecting suspicious activity.
|
|
|
|
Args:
|
|
site_id_list: List of site IDs from SiteList. Pass SPECIFIC sites; Empty = ALL sites
|
|
(aggregates every site's logs and can be very large - prefer targeted queries).
|
|
|
|
Workflow: SiteList -> GetSitesLogs (takes id; resolves domain internally).
|
|
"""
|
|
import json
|
|
from logsModelV2.siteModel import main
|
|
logs_model = main()
|
|
|
|
if not site_id_list:
|
|
all_sites = public.M('sites').field('id,name').select()
|
|
if not all_sites:
|
|
return _xml_response("GetSitesLogs", "done",
|
|
json.dumps({"error": "No sites found in panel."}, ensure_ascii=False, indent=2))
|
|
site_id_list = [s['id'] for s in all_sites]
|
|
|
|
results = {}
|
|
last_site_name = None
|
|
for sid in site_id_list:
|
|
site_data = public.M('sites').field('name').where("id=?", int(sid)).find()
|
|
if not site_data:
|
|
results[str(sid)] = {"error": "Site not found in panel."}
|
|
continue
|
|
site_name = site_data['name']
|
|
last_site_name = site_name
|
|
logs = logs_model.GetSiteLogs(public.to_dict_obj({"siteName": site_name})).get("message")
|
|
results[site_name] = logs
|
|
|
|
if len(site_id_list) == 1 and last_site_name in results:
|
|
return _xml_response("GetSitesLogs", "done",
|
|
json.dumps(results[last_site_name], ensure_ascii=False, indent=2))
|
|
return _xml_response("GetSitesLogs", "done", json.dumps(results, ensure_ascii=False, indent=2))
|
|
|
|
|
|
@register_tool(category="Website", name_cn="Get Website Traffic Data", risk_level="low")
|
|
def GetSiteOverview(site_id_list: List[int] = None, sort_by: str = "") -> str:
|
|
"""
|
|
Get 7-day traffic statistics (UV, PV, bandwidth) for websites.
|
|
|
|
Args:
|
|
site_id_list: List of site IDs from SiteList. Pass SPECIFIC sites; Empty = ALL sites
|
|
(can be large with many sites - prefer targeted queries, or use sort_by="traffic" for a ranked overview).
|
|
sort_by: Sort field. Optional. "traffic" = sort by traffic desc. Empty = no sort.
|
|
|
|
Returns: Per-site traffic data. When sort_by="traffic", returns ranked list for all sites.
|
|
|
|
Workflow: SiteList -> GetSiteOverview (takes id; resolves domain internally).
|
|
"""
|
|
import json
|
|
from projectModelV2.monitorModel import main as monitor
|
|
|
|
# 排序模式：直接返回聚合排名数据 (不依赖站点列表)
|
|
if sort_by == "traffic":
|
|
monitordata = monitor().get_overview(public.to_dict_obj({"metric": "traffic", "order": "desc"})).get("message")
|
|
return _xml_response("GetSiteOverview", "done", json.dumps(monitordata, ensure_ascii=False, indent=2))
|
|
|
|
if not site_id_list:
|
|
all_sites = public.M('sites').field('id,name').select()
|
|
if not all_sites:
|
|
return _xml_response("GetSiteOverview", "done",
|
|
json.dumps({"error": "No sites found in panel."}, ensure_ascii=False, indent=2))
|
|
site_id_list = [s['id'] for s in all_sites]
|
|
|
|
results = {}
|
|
last_site_name = None
|
|
for sid in site_id_list:
|
|
site_data = public.M('sites').field('name').where("id=?", int(sid)).find()
|
|
if not site_data:
|
|
results[str(sid)] = {"error": "Site not found in panel."}
|
|
continue
|
|
site_name = site_data['name']
|
|
last_site_name = site_name
|
|
monitordata = monitor().get_overview(public.to_dict_obj({"site_name": site_name})).get("message")
|
|
results[site_name] = monitordata
|
|
|
|
if len(site_id_list) == 1 and last_site_name in results:
|
|
return _xml_response("GetSiteOverview", "done",
|
|
json.dumps(results[last_site_name], ensure_ascii=False, indent=2))
|
|
return _xml_response("GetSiteOverview", "done", json.dumps(results, ensure_ascii=False, indent=2))
|
|
|
|
|
|
# @register_tool(category="Website", name_cn="Backup Website", risk_level="high")
|
|
# def WebSiteBackup(site_ids: List[int]) -> str:
|
|
# """
|
|
# Backup websites by site ID. Auto-detects type: PHP (files only), WP/WP2 (files+DB), Node.js (files).
|
|
#
|
|
# Prefer this over manual `RunCommand tar` — it detects site type and records backups in the panel for one-click restore.
|
|
#
|
|
# Args:
|
|
# site_ids: List of site IDs from SiteList. Required. e.g. [1, 2, 3].
|
|
#
|
|
# Returns: Per-site backup trigger result. Backup runs asynchronously.
|
|
#
|
|
# Workflow: SiteList -> WebSiteBackup.
|
|
# """
|
|
# import json
|
|
#
|
|
# if not site_ids:
|
|
# return _xml_response("WebSiteBackup", "error", "No site IDs provided. Use SiteList to list available sites.")
|
|
#
|
|
# sids = [int(sid) for sid in site_ids]
|
|
# site_results = [] # 每个站点的预检信息
|
|
#
|
|
# # 1. 逐个查询站点信息，按类型分组
|
|
# php_ids, wp_ids, node_ids = [], [], []
|
|
# for sid in sids:
|
|
# site_info = public.M('sites').where('id=?', (sid,)).field('id,name,project_type').find()
|
|
# if not site_info:
|
|
# site_results.append({"id": sid, "status": "error", "msg": "Site not found"})
|
|
# continue
|
|
#
|
|
# ptype = (site_info.get('project_type') or '').upper()
|
|
# entry = {"id": sid, "name": site_info.get('name', ''), "type": ptype}
|
|
#
|
|
# if ptype == 'PHP':
|
|
# entry["backup_mode"] = "file-only"
|
|
# php_ids.append(sid)
|
|
# elif ptype in ('WP', 'WP2'):
|
|
# entry["backup_mode"] = "full"
|
|
# wp_ids.append(sid)
|
|
# elif ptype == 'NODE':
|
|
# entry["backup_mode"] = "file-only"
|
|
# node_ids.append(sid)
|
|
# else:
|
|
# entry["status"] = "error"
|
|
# entry["msg"] = "Unsupported type '{}'. Supported: PHP, WP/WP2, Node.".format(ptype)
|
|
#
|
|
# site_results.append(entry)
|
|
#
|
|
# # 2. 按类型各调一次备份接口
|
|
# triggered = {"php": False, "wp": False, "node": False}
|
|
#
|
|
# if php_ids:
|
|
# try:
|
|
# from panel_site_v2 import panelSite
|
|
# args = public.to_dict_obj({'id': php_ids, 'bak_type': 0, 'site_type': 'php'})
|
|
# res = panelSite().ai_php_wp_backup(args)
|
|
# triggered["php"] = isinstance(res, dict) and res.get('status') == 0
|
|
# except Exception:
|
|
# pass
|
|
#
|
|
# if wp_ids:
|
|
# try:
|
|
# from panel_site_v2 import panelSite
|
|
# args = public.to_dict_obj({'id': wp_ids, 'bak_type': 3, 'site_type': 'wp'})
|
|
# res = panelSite().ai_php_wp_backup(args)
|
|
# triggered["wp"] = isinstance(res, dict) and res.get('status') == 0
|
|
# except Exception:
|
|
# pass
|
|
#
|
|
# if node_ids:
|
|
# try:
|
|
# from mod.project.nodejs.comMod import main as nodejs_main
|
|
# args = public.to_dict_obj({'id': node_ids, 'backup_type': 0})
|
|
# res = nodejs_main().nodejs_backup(args)
|
|
# triggered["node"] = isinstance(res, dict) and res.get('status') == 0
|
|
# except Exception:
|
|
# pass
|
|
#
|
|
# # 3. 标记各站点触发结果
|
|
# for entry in site_results:
|
|
# if entry.get('status') == 'error':
|
|
# continue # 已标记错误的跳过
|
|
# ptype = entry.get('type', '')
|
|
# if ptype == 'PHP':
|
|
# entry["status"] = "ok" if triggered["php"] else "error"
|
|
# if not triggered["php"]:
|
|
# entry["msg"] = "PHP backup trigger failed"
|
|
# elif ptype in ('WP', 'WP2'):
|
|
# entry["status"] = "ok" if triggered["wp"] else "error"
|
|
# if not triggered["wp"]:
|
|
# entry["msg"] = "WordPress backup trigger failed"
|
|
# elif ptype == 'NODE':
|
|
# entry["status"] = "ok" if triggered["node"] else "error"
|
|
# if not triggered["node"]:
|
|
# entry["msg"] = "Node.js backup trigger failed"
|
|
#
|
|
# success = sum(1 for r in site_results if r.get('status') == 'ok')
|
|
# failed = sum(1 for r in site_results if r.get('status') == 'error')
|
|
#
|
|
# output = {
|
|
# "total": len(sids),
|
|
# "success": success,
|
|
# "failed": failed,
|
|
# "results": site_results,
|
|
# "progress_hint": "Backup is running asynchronously. Check progress in the backup log.",
|
|
# }
|
|
#
|
|
# status = "done" if failed == 0 else ("partial" if success > 0 else "error")
|
|
# return _xml_response("WebSiteBackup", status, json.dumps(output, ensure_ascii=False, indent=2))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Database Management
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="Database", name_cn="List Databases", risk_level="low")
|
|
def DBList(limit: int = 200) -> str:
|
|
"""
|
|
List all databases with id, pid (parent site id, 0 = standalone), name, username, type,
|
|
allowed IP (accept), sid (server id, 0 = local MySQL), conn_config (connection params:
|
|
host/port/db_host etc), and backup_count.
|
|
|
|
Returns: JSON array. Use id for DBDelete.
|
|
|
|
Workflow: DBList -> DBCreate / DBDelete / DBChangePassword.
|
|
|
|
Args:
|
|
limit: Max rows to return (default 200). Rows beyond are omitted.
|
|
"""
|
|
try:
|
|
rows = public.M('databases').field('id,pid,name,username,accept,type,sid,conn_config').select()
|
|
# 批量备份计数 (databases -> type='1', 一次 where_in)
|
|
bak_map = {}
|
|
try:
|
|
ids = [r['id'] for r in rows]
|
|
if ids:
|
|
bak_rows = public.S('backup').where_in('pid', ids).where("type", "1").group('pid').field(
|
|
'pid', 'count(*) as cnt').select()
|
|
bak_map = {b['pid']: b['cnt'] for b in bak_rows}
|
|
except Exception:
|
|
pass
|
|
for r in rows:
|
|
try:
|
|
r['conn_config'] = json.loads(r['conn_config']) if r.get('conn_config') else {}
|
|
except Exception:
|
|
r['conn_config'] = {}
|
|
r['backup_count'] = bak_map.get(r['id'], 0)
|
|
rows = _truncate_result(rows, limit)
|
|
return _xml_response("DBList", "done", json.dumps(rows, ensure_ascii=False, default=str))
|
|
except Exception as e:
|
|
return _xml_response("DBList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Database", name_cn="Create Database", risk_level="medium")
|
|
def DBCreate(name: str, password: str, db_user: str = "", ps: str = "") -> str:
|
|
"""
|
|
Create a MySQL database (utf8mb4).
|
|
|
|
Args:
|
|
name: Database name. Required.
|
|
password: Database password. Required.
|
|
db_user: Database username. Optional, defaults to same as name.
|
|
ps: Remark. Optional.
|
|
|
|
Workflow: DBList -> DBCreate.
|
|
|
|
Note: Remote access is enabled by default (address="%").
|
|
"""
|
|
try:
|
|
from class_v2.database_v2 import database
|
|
params = {
|
|
"name": name, "password": password,
|
|
"db_user": db_user or name,
|
|
"codeing": "utf8mb4", "sid": 0, "active": True,
|
|
"address": "%", "ps": ps, "dtype": "MySQL",
|
|
}
|
|
data = database().AddDatabase(_to_obj(params))
|
|
return _xml_response("DBCreate", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DBCreate", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Database", name_cn="Delete Database", risk_level="high")
|
|
def DBDelete(id: int, name: str) -> str:
|
|
"""
|
|
Delete a database. Irreversible.
|
|
|
|
Args:
|
|
id: Database ID from DBList. Required.
|
|
name: Database name. Required.
|
|
|
|
Workflow: DBList -> DBDelete.
|
|
"""
|
|
try:
|
|
from class_v2.database_v2 import database
|
|
data = database().DeleteDatabase(_to_obj({"id": id, "name": name}))
|
|
return _xml_response("DBDelete", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DBDelete", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Database", name_cn="Change Database Password", risk_level="high")
|
|
def DBChangePassword(id: int, db_user: str, password: str) -> str:
|
|
"""
|
|
Change the password of a panel-managed database user. Syncs BOTH MySQL and panel records.
|
|
|
|
Args:
|
|
id: Database id from DBList. Required.
|
|
db_user: Database username (DBList 'username' field). Required.
|
|
password: New password (no Chinese; limited special chars). Required.
|
|
|
|
Workflow: DBList -> DBChangePassword.
|
|
|
|
Note: This is the ONLY safe way to change a DB password - it keeps MySQL and panel
|
|
records in sync. NEVER use MysqlExecute to alter user passwords.
|
|
"""
|
|
try:
|
|
from class_v2.database_v2 import database
|
|
data = database().ResDatabasePassword(
|
|
_to_obj({"id": id, "name": db_user, "password": password}))
|
|
return _xml_response("DBChangePassword", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DBChangePassword", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Database", name_cn="Backup Database", risk_level="high")
|
|
def DBBackup(db_id: int) -> str:
|
|
"""
|
|
Back up a MySQL database (full dump to a timestamped .sql.zip). Blocks until the dump finishes; slow on large DBs.
|
|
|
|
Args:
|
|
db_id: Database ID from DBList (MySQL only). Required.
|
|
|
|
Workflow: DBList -> DBBackup -> DBBackupList.
|
|
|
|
Prefer this over raw mysqldump — it records the backup in the panel registry.
|
|
"""
|
|
try:
|
|
from class_v2.database_v2 import database
|
|
data = database().ToBackup(_to_obj({"id": db_id}))
|
|
return _xml_response("DBBackup", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DBBackup", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Database", name_cn="List Database Backups", risk_level="low")
|
|
def DBBackupList(search: str = "", limit: int = 100) -> str:
|
|
"""
|
|
(read-only) List MySQL database backup files across all databases, newest first.
|
|
|
|
Args:
|
|
search: Optional filename keyword to filter.
|
|
limit: Max items (default 100).
|
|
|
|
Workflow: DBBackup -> DBBackupList.
|
|
"""
|
|
try:
|
|
from class_v2.database_v2 import database
|
|
params = {"p": 1, "limit": limit}
|
|
if search:
|
|
params["search"] = search
|
|
data = database().GetBackup(_to_obj(params))
|
|
return _xml_response("DBBackupList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DBBackupList", "error", str(e))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# SSL Certificate Management
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="SSL", name_cn="List SSL Certificates", risk_level="low")
|
|
def SSLList(search: str = "", limit: int = 100) -> str:
|
|
"""
|
|
List SSL certificates with domain, expiry, renewal status, and hash.
|
|
|
|
Returns: JSON array. hash identifies each certificate.
|
|
|
|
Args:
|
|
search: Optional domain keyword to filter results.
|
|
limit: Max items to return (default 100). Items beyond are omitted.
|
|
"""
|
|
try:
|
|
from class_v2.ssl_domainModelV2.api import DomainObject
|
|
params = {"search": search} if search else {}
|
|
data = DomainObject().list_ssl_info(_to_obj(params))
|
|
data = _truncate_result(data, limit)
|
|
return _xml_response("SSLList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("SSLList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Website", name_cn="Auto Apply SSL", risk_level="high")
|
|
def AutoSSL(site_id: int) -> str:
|
|
"""
|
|
Auto-apply free SSL for a site (Let's Encrypt). Reuses existing valid certs if matched.
|
|
|
|
Note: Let's Encrypt SSL can be applied using the server's public IP.
|
|
|
|
Prefer this over `RunCommand certbot` — it reuses panel DNS accounts and records certs under panel management.
|
|
|
|
Verification method is auto-decided: all domains have a DNS Provider -> DNS-01; otherwise HTTP-01 (wildcard domains require DNS-01). Blocks until the certificate is applied or fails, then reports the final result.
|
|
|
|
Args:
|
|
site_id: Site ID from SiteList. Required.
|
|
|
|
Returns: Pre-check (site info, domain list, DNS resolution) plus the final apply result in ssl_result ({"status": bool, "msg": str}).
|
|
|
|
Workflow: SiteList -> AutoSSL.
|
|
"""
|
|
import json
|
|
import socket
|
|
|
|
sid = int(site_id)
|
|
precheck = {}
|
|
|
|
# 1. 查询站点信息
|
|
try:
|
|
precheck['site_info'] = public.M('sites').where('id=?', sid).field('id,name,project_type').find()
|
|
except Exception as e:
|
|
precheck['site_info'] = {'error': str(e)}
|
|
|
|
# 2. 查询域名列表
|
|
domain_names = []
|
|
try:
|
|
domain_records = public.M('domain').where('pid=?', sid).field('name').select()
|
|
if domain_records and isinstance(domain_records, list):
|
|
domain_names = [d.get('name', '') for d in domain_records if d.get('name')]
|
|
precheck['domains'] = domain_names
|
|
except Exception as e:
|
|
precheck['domains'] = {'error': str(e)}
|
|
|
|
# 3. A记录解析查询 (公共DNS + 本地DNS, 单域名超时3秒)
|
|
dns_results = {}
|
|
for domain in domain_names:
|
|
record = {}
|
|
# 公共DNS查询
|
|
try:
|
|
results = socket.getaddrinfo(domain, None, socket.AF_INET, socket.SOCK_STREAM)
|
|
record['public_dns'] = sorted(set(r[4][0] for r in results))
|
|
except Exception:
|
|
record['public_dns'] = lang('unresolved')
|
|
# 本地DNS查询
|
|
try:
|
|
p = subprocess.run(
|
|
['dig', '@127.0.0.1', domain, '+short', '+time=3'],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
ips = sorted(set(
|
|
line.strip() for line in p.stdout.strip().splitlines() if line.strip() and not line.startswith(';')))
|
|
record['local_dns'] = ips if ips else lang('unresolved')
|
|
except Exception:
|
|
record['local_dns'] = lang('unresolved')
|
|
dns_results[domain] = record
|
|
precheck['dns_resolution'] = dns_results
|
|
|
|
# 同步申请SSL, 阻塞至完成
|
|
try:
|
|
from ssl_domainModelV2.service import apply_site_ssl_sync
|
|
precheck['ssl_result'] = apply_site_ssl_sync(sid)
|
|
except Exception as e:
|
|
precheck['ssl_result'] = {"status": False, "msg": f"SSL apply error: {e}"}
|
|
|
|
return _xml_response("AutoSSL", "done", json.dumps(precheck, ensure_ascii=False, indent=2))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# DNS Provider & Records
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="DNS", name_cn="List DNS Providers", risk_level="low")
|
|
def DNSProviderList() -> str:
|
|
"""
|
|
List DNS provider accounts with id, name, and alias.
|
|
|
|
Returns: JSON array. id is used as pid/search_pid in all DNS record operations. Each entry also includes a 'domains' field.
|
|
|
|
aaPanelDns is the panel's self-hosted DNS (powered by PowerDNS, zones in BIND9 format, no API credentials); other providers are third-party APIs.
|
|
"""
|
|
try:
|
|
from class_v2.ssl_domainModelV2.api import DomainObject
|
|
data = DomainObject().list_dns_api(_to_obj({}))
|
|
try:
|
|
for d in data.get('message', {}).get('data', []):
|
|
if 'api_key' in d:
|
|
d.pop('api_key') # Remove sensitive API keys from output
|
|
except:
|
|
pass
|
|
return _xml_response("DNSProviderList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DNSProviderList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="DNS", name_cn="List DNS Records", risk_level="low")
|
|
def DNSRecordList(search_pid: int, domain: str = "", search: str = "") -> str:
|
|
"""
|
|
List DNS records under a provider.
|
|
|
|
Args:
|
|
search_pid: Provider ID from DNSProviderList.
|
|
domain: Filter by domain. Optional.
|
|
search: Filter by keyword. Optional.
|
|
|
|
Workflow: DNSProviderList -> DNSRecordList -> DNSRecordDelete.
|
|
"""
|
|
try:
|
|
from class_v2.ssl_domainModelV2.api import DomainObject
|
|
params = {"search_pid": search_pid}
|
|
if domain:
|
|
params["domain"] = domain
|
|
if search:
|
|
params["search"] = search
|
|
data = DomainObject().list_dns_record(_to_obj(params))
|
|
return _xml_response("DNSRecordList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DNSRecordList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="DNS", name_cn="Create DNS Record", risk_level="high")
|
|
def DNSRecordCreate(pid: int, domain: str, record: str, record_type: str,
|
|
record_value: str, ttl: int = 600, priority: int = 0,
|
|
proxy: int = 0, ps: str = "") -> str:
|
|
"""
|
|
Create a DNS record under a configured DNS provider.
|
|
|
|
Args:
|
|
pid: Provider ID from DNSProviderList.
|
|
domain: Domain name. e.g. "example.com".
|
|
record: Record name. e.g. "www"/"@"(root)/"*"(wildcard).
|
|
record_type: Record type. e.g. "A"/"AAAA"/"CNAME"/"MX"/"TXT"/"SRV".
|
|
record_value: Record value. e.g. "1.2.3.4" for A record.
|
|
ttl: TTL in seconds. Default 600.
|
|
priority: MX/SRV priority. Default 0.
|
|
proxy: Cloudflare CDN proxy only (ignored by other providers). 0=off, 1=on. Default 0.
|
|
ps: Remark. Optional.
|
|
|
|
Workflow: DNSProviderList -> DNSRecordList -> DNSRecordCreate.
|
|
"""
|
|
try:
|
|
from class_v2.ssl_domainModelV2.api import DomainObject
|
|
params = {
|
|
"pid": pid, "domain": domain, "record": record,
|
|
"record_type": record_type, "record_value": record_value,
|
|
"ttl": ttl, "priority": priority, "proxy": proxy,
|
|
}
|
|
if ps:
|
|
params["ps"] = ps
|
|
data = DomainObject().create_dns_record(_to_obj(params))
|
|
return _xml_response("DNSRecordCreate", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DNSRecordCreate", "error", str(e))
|
|
|
|
|
|
@register_tool(category="DNS", name_cn="Delete DNS Record", risk_level="high")
|
|
def DNSRecordDelete(id: int) -> str:
|
|
"""
|
|
Delete a DNS record by ID. Irreversible.
|
|
|
|
Args:
|
|
id: Record ID from DNSRecordList.
|
|
|
|
Workflow: DNSRecordList -> DNSRecordDelete.
|
|
"""
|
|
try:
|
|
from class_v2.ssl_domainModelV2.api import DomainObject
|
|
data = DomainObject().delete_dns_record(_to_obj({"id": id}))
|
|
return _xml_response("DNSRecordDelete", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("DNSRecordDelete", "error", str(e))
|
|
|
|
|
|
# @register_tool(category="DNS", name_cn="Edit DNS Record", risk_level="high")
|
|
# def DNSRecordEdit(id: int, pid: int, domain: str, record: str, record_type: str,
|
|
# record_value: str, ttl: int = 600, proxy: int = 0,
|
|
# priority: int = 0, ps: str = "") -> str:
|
|
# """
|
|
# Edit an existing DNS record.
|
|
#
|
|
# Args:
|
|
# id: Record ID from DNSRecordList. Required.
|
|
# pid: Provider ID from DNSProviderList. Required.
|
|
# domain: Domain name. Required.
|
|
# record: Record name. Required.
|
|
# record_type: Type. Required. e.g. "A"/"CNAME"/"MX"/"TXT".
|
|
# record_value: New value. Required.
|
|
# ttl: TTL in seconds. Default 600.
|
|
# proxy: CloudFlare CDN proxy. 0=off, 1=on. Default 0.
|
|
# priority: MX/SRV priority. Default 0.
|
|
# ps: Remark. Optional.
|
|
#
|
|
# Workflow: DNSRecordList -> DNSRecordEdit (get current values first).
|
|
# """
|
|
# try:
|
|
# from class_v2.ssl_domainModelV2.api import DomainObject
|
|
# params = {
|
|
# "id": id, "pid": pid, "domain": domain, "record": record,
|
|
# "record_type": record_type, "record_value": record_value,
|
|
# "ttl": ttl, "proxy": proxy, "priority": priority, "ps": ps,
|
|
# }
|
|
# data = DomainObject().edit_dns_record(_to_obj(params))
|
|
# return _xml_response("DNSRecordEdit", "done", json.dumps(data, ensure_ascii=False))
|
|
# except Exception as e:
|
|
# return _xml_response("DNSRecordEdit", "error", str(e))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Plugin & Software Management
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
# @register_tool(category="Plugin", name_cn="List Software Store", risk_level="low")
|
|
# def PluginSoftList(type: str = "0", query: str = "", p: int = 1, limit: int = 15) -> str:
|
|
# """
|
|
# List software in the aaPanel app store. Supports type and keyword filtering.
|
|
#
|
|
# Args:
|
|
# #type: "0" = all (default), "-1" = installed only.
|
|
# query: Search keyword. e.g. "nginx"/"docker"/"php". Optional.
|
|
# p: Page number. Default 1.
|
|
# limit: Items per page. Default 15.
|
|
#
|
|
# Returns: Paginated list with versions info. Extract sName and m_version for PluginInstall.
|
|
#
|
|
# Workflow: PluginSoftList -> PluginInstall.
|
|
# """
|
|
# try:
|
|
# from class_v2.panel_plugin_v2 import panelPlugin
|
|
# params = {"type": type, "p": p, "limit": limit}
|
|
# if query:
|
|
# params["query"] = query
|
|
# data = panelPlugin().get_soft_list(_to_obj(params))
|
|
# return _xml_response("PluginSoftList", "done", json.dumps(data, ensure_ascii=False))
|
|
# except Exception as e:
|
|
# return _xml_response("PluginSoftList", "error", str(e))
|
|
#
|
|
#
|
|
# @register_tool(category="Plugin", name_cn="Install Software", risk_level="high")
|
|
# def PluginInstall(sName: str, version: str) -> str:
|
|
# """
|
|
# Install a software package from the app store. This is an ASYNC trigger.
|
|
#
|
|
# Args:
|
|
# sName: Software name from PluginSoftList. Required.
|
|
# version: m_version from PluginSoftList. Required. e.g. "1.24"/"8.2"/"5.7".
|
|
#
|
|
# Workflow: PluginSoftList -> PluginInstall.
|
|
#
|
|
# IMPORTANT: This is an asynchronous operation. The install runs in the background.
|
|
# This tool ONLY triggers the install and returns immediately.
|
|
# NEVER poll, wait, retry, or check install progress after calling this tool.
|
|
# Just report the trigger result (success/failure) to the user and stop.
|
|
# """
|
|
# try:
|
|
# from class_v2.panel_plugin_v2 import panelPlugin
|
|
# params = {"sName": sName, "version": version}
|
|
# data = panelPlugin().install_plugin(_to_obj(params))
|
|
# return _xml_response("PluginInstall", "done", json.dumps(data, ensure_ascii=False))
|
|
# except Exception as e:
|
|
# return _xml_response("PluginInstall", "error", str(e))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Service Management
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def _service_op(service: str, action: str) -> str:
|
|
"""管理面板服务: 主路 system().ServiceAdmin (走 /etc/init.d/<service>);
|
|
主路失败 (返回 status!=0 或抛异常) 才 fallback 到 systemctl, 用 is-active 验证成败.
|
|
|
|
action: "restart" 或 "stop". 模块级唯一对外入口, 内部封装 fallback 细节.
|
|
"""
|
|
tool = "ServiceRestart" if action == "restart" else "ServiceStop"
|
|
|
|
def _msg(data) -> str:
|
|
# 从 return_message 结构 ({status, message:{result}}) 提取可读错误信息
|
|
if isinstance(data, dict):
|
|
msg = data.get("message")
|
|
if isinstance(msg, dict) and "result" in msg:
|
|
return str(msg["result"])
|
|
return json.dumps(data, ensure_ascii=False)
|
|
return str(data)
|
|
|
|
def _run_systemctl(svc: str, act: str):
|
|
# 执行 systemctl <act> <svc>, 用 is-active 验证. 返回 (ok, detail)
|
|
try:
|
|
public.ExecShell("systemctl {} {}".format(act, svc))
|
|
except Exception as e:
|
|
return False, "exec error: " + str(e)
|
|
res = public.ExecShell("systemctl is-active {}".format(svc))
|
|
if isinstance(res, tuple):
|
|
state = (res[0] or "").strip()
|
|
elif isinstance(res, str):
|
|
state = res.strip()
|
|
else:
|
|
state = ""
|
|
if act == "restart":
|
|
return state == "active", "is-active={}".format(state)
|
|
return state in ("inactive", "failed", "deactivating"), "is-active={}".format(state)
|
|
|
|
primary_msg = None
|
|
# 主路: ServiceAdmin 优先, 保持当前工具行为
|
|
try:
|
|
from class_v2.system_v2 import system
|
|
data = system().ServiceAdmin(_to_obj({"name": service, "type": action}))
|
|
time.sleep(1)
|
|
if isinstance(data, dict) and data.get("status") == 0:
|
|
return _xml_response(tool, "done", json.dumps(data, ensure_ascii=False))
|
|
primary_msg = _msg(data)
|
|
except Exception as e:
|
|
primary_msg = str(e)
|
|
|
|
# 兜底: systemctl + is-active 验证
|
|
ok, detail = _run_systemctl(service, action)
|
|
time.sleep(1)
|
|
if ok:
|
|
return _xml_response(tool, "done", json.dumps(
|
|
{"status": 0, "fallback": "systemctl", "primary_error": primary_msg, "detail": detail},
|
|
ensure_ascii=False))
|
|
return _xml_response(tool, "error",
|
|
"ServiceAdmin failed [{}]; systemctl fallback failed: {}".format(primary_msg, detail))
|
|
|
|
|
|
@register_tool(category="Service", name_cn="Get Service Status", risk_level="low")
|
|
def ServiceStatus(name_list: List[str]) -> str:
|
|
"""
|
|
Query install/running status and version for one or more services.
|
|
|
|
Args:
|
|
name_list: Service names. Required. e.g. ["nginx", "php-74", "mysql"].
|
|
|
|
Returns: One line per service, e.g. "nginx: installed, running, v1.24" or "mysql: not installed".
|
|
|
|
Workflow: ServiceStatus -> ServiceRestart / ServiceStop.
|
|
"""
|
|
try:
|
|
names = name_list if isinstance(name_list, list) else [name_list]
|
|
report = _ServiceStatusEnricher.query(names)
|
|
return _xml_response("ServiceStatus", "done", report)
|
|
except Exception as e:
|
|
return _xml_response("ServiceStatus", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Service", name_cn="Restart Service", risk_level="high")
|
|
def ServiceRestart(service: str) -> str:
|
|
"""
|
|
Restart a panel-managed service. Use with caution on production servers.
|
|
|
|
Routes to ServiceAdmin (/etc/init.d/<service>) first; on failure it falls back to `systemctl restart` and confirms via `systemctl is-active`.
|
|
|
|
Prefer this over `RunCommand systemctl restart` for panel-managed services — it keeps panel state in sync.
|
|
|
|
Args:
|
|
service: Panel-managed service name. Required. e.g. "nginx"/"mysql"/"apache"/"php-fpm".
|
|
|
|
Workflow: ServiceStatus -> ServiceRestart.
|
|
"""
|
|
return _service_op(service, "restart")
|
|
|
|
|
|
@register_tool(category="Service", name_cn="Stop Service", risk_level="high")
|
|
def ServiceStop(service: str) -> str:
|
|
"""
|
|
Stop a panel-managed service. Related functionality will become unavailable.
|
|
|
|
Routes to ServiceAdmin (/etc/init.d/<service>) first; on failure it falls back to `systemctl stop` and confirms via `systemctl is-active`.
|
|
|
|
Prefer this over `RunCommand systemctl stop` for panel-managed services — it keeps panel state in sync.
|
|
|
|
Args:
|
|
service: Panel-managed service name. Required. e.g. "nginx"/"mysql"/"apache"/"php-fpm".
|
|
|
|
Workflow: ServiceStatus -> ServiceStop.
|
|
"""
|
|
return _service_op(service, "stop")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# FTP Account Management
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="FTP", name_cn="List FTP Accounts", risk_level="low")
|
|
def FTPList() -> str:
|
|
"""
|
|
List FTP accounts with id, name, path, status.
|
|
|
|
Returns: JSON array. Use id for management.
|
|
|
|
Workflow: FTPList -> FTPCreate.
|
|
"""
|
|
try:
|
|
rows = public.M('ftps').field('id,name,path,status,ps,addtime').select()
|
|
return _xml_response("FTPList", "done", json.dumps(rows, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FTPList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="FTP", name_cn="Create FTP Account", risk_level="medium")
|
|
def FTPCreate(ftp_username: str, ftp_password: str, path: str, ps: str = "") -> str:
|
|
"""
|
|
Create an FTP account.
|
|
|
|
Args:
|
|
ftp_username: Username. Required. Min 3 chars, no special characters.
|
|
ftp_password: Password. Required. Min 6 chars.
|
|
path: Root directory. Required. e.g. "/www/wwwroot/ftptest".
|
|
ps: Remark. Optional.
|
|
|
|
Workflow: FTPList -> FTPCreate.
|
|
"""
|
|
try:
|
|
from class_v2.ftp_v2 import ftp
|
|
params = {"ftp_username": ftp_username, "ftp_password": ftp_password, "path": path, "ps": ps}
|
|
data = ftp().AddUser(_to_obj(params))
|
|
return _xml_response("FTPCreate", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FTPCreate", "error", str(e))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Cron Job Management
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="Cron", name_cn="List Cron Jobs", risk_level="low")
|
|
def CronList(limit: int = 100) -> str:
|
|
"""
|
|
List cron jobs with id, name, type, schedule, status, and command.
|
|
|
|
Returns: JSON array. Use id for CronDelete.
|
|
|
|
Workflow: CronList -> CronCreate / CronDelete.
|
|
|
|
Args:
|
|
limit: Max items to return (default 100). Items beyond are omitted.
|
|
"""
|
|
try:
|
|
from class_v2.crontab_v2 import crontab
|
|
data = crontab().GetCrontab(_to_obj({}))
|
|
data = _truncate_result(data, limit)
|
|
return _xml_response("CronList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("CronList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Cron", name_cn="Create Cron Job", risk_level="medium")
|
|
def CronCreate(name: str, type: str, where1: str, sBody: str, sType: str = "toShell",
|
|
hour: int = 0, minute: int = 0, user: str = "root",
|
|
save: str = "", backupTo: str = "", sName: str = "",
|
|
urladdress: str = "", db_type: str = "", notice: int = 0,
|
|
notice_channel: str = "", save_local: int = 0) -> str:
|
|
"""
|
|
Create a scheduled task.
|
|
|
|
Args:
|
|
name: Job name. Required. e.g. "Daily Backup".
|
|
type: Schedule type. Required.
|
|
"minute-n" -> where1: "*/5" (every N minutes)
|
|
"hour" -> where1: "1" (at minute N of each hour)
|
|
"hour-n" -> where1: "2" (every N hours)
|
|
"day" -> where1: "" (daily, use hour+minute for time)
|
|
"day-n" -> where1: "3" (every N days)
|
|
"week" -> where1: "1" (day of week 0=Sun-6=Sat, use hour+minute for time)
|
|
"month" -> where1: "15" (day of month, use hour+minute for time)
|
|
"once" -> where1: "2025-12-31 01:30" (one-time)
|
|
where1: Cycle/frequency value. Required. Format depends on type (see above).
|
|
sBody: Shell command or script content. Required. Will be executed by cron when sType="toShell".
|
|
sType: Task type. "toShell" (shell command), "toUrl" (request URL), "database"/"site"/"path" (backup), "logs"/"webshell"/"sync_time"/"startup_services" etc.
|
|
hour: Execution hour for day/week/month types. Default 0.
|
|
minute: Execution minute for day/week/month/hour types. Default 0.
|
|
user: Execute user. Default "root".
|
|
save: Backup retention count. Optional.
|
|
backupTo: Backup storage location. Optional.
|
|
sName: Backup scope for database/site type. Optional.
|
|
urladdress: URL for toUrl type. Optional.
|
|
db_type: Database type for database backup. Optional.
|
|
notice: Enable notification. 0=off, 1=on. Default 0.
|
|
notice_channel: Notification channel. Optional.
|
|
save_local: Keep local copy. 0=off, 1=on. Default 0.
|
|
|
|
Workflow: CronList -> CronCreate.
|
|
"""
|
|
try:
|
|
from class_v2.crontab_v2 import crontab
|
|
params = {
|
|
"name": name, "type": type, "where1": where1, "sBody": sBody, "sType": sType,
|
|
"hour": hour, "minute": minute, "user": user,
|
|
"save": save, "backupTo": backupTo, "sName": sName,
|
|
"urladdress": urladdress, "db_type": db_type,
|
|
"notice": notice, "notice_channel": notice_channel, "save_local": save_local,
|
|
}
|
|
data = crontab().AddCrontab(_to_obj(params))
|
|
return _xml_response("CronCreate", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("CronCreate", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Cron", name_cn="Delete Cron Job", risk_level="high")
|
|
def CronDelete(id: int) -> str:
|
|
"""
|
|
Delete a cron job. Irreversible.
|
|
|
|
Args:
|
|
id: Job ID from CronList. Required.
|
|
|
|
Workflow: CronList -> CronDelete.
|
|
"""
|
|
try:
|
|
from class_v2.crontab_v2 import crontab
|
|
data = crontab().DelCrontab(_to_obj({"id": id}))
|
|
return _xml_response("CronDelete", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("CronDelete", "error", str(e))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# SSH Security
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="SSH", name_cn="SSH Login Logs", risk_level="low")
|
|
def SSHLogList(limit: int = 100) -> str:
|
|
"""
|
|
Get SSH login history: login time, IP, username, and result.
|
|
|
|
Returns: JSON array of SSH login records.
|
|
|
|
Workflow: SSHLogList -> SSHIntrusionList (if anomalies detected).
|
|
|
|
Args:
|
|
limit: Max records to return (default 100). Records beyond are omitted.
|
|
"""
|
|
try:
|
|
from mod.modController import Controller
|
|
args = _to_obj({"mod_name": "ssh", "sub_mod_name": "com", "def_name": "get_ssh_list", "data": "{}"})
|
|
data = Controller().model(args)
|
|
data = _truncate_result(data, limit)
|
|
return _xml_response("SSHLogList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("SSHLogList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="SSH", name_cn="SSH Intrusion Detection", risk_level="low")
|
|
def SSHIntrusionList(limit: int = 100) -> str:
|
|
"""
|
|
Get SSH intrusion detection records: brute-force attempts, anomalous logins.
|
|
|
|
Returns: JSON array of intrusion events.
|
|
|
|
Workflow: SSHLogList -> SSHIntrusionList -> FirewallPortRuleList / FirewallIPRuleList (block suspicious IPs).
|
|
|
|
Args:
|
|
limit: Max records to return (default 100). Records beyond are omitted.
|
|
"""
|
|
try:
|
|
from mod.modController import Controller
|
|
args = _to_obj({"mod_name": "ssh", "sub_mod_name": "com", "def_name": "get_ssh_intrusion", "data": "{}"})
|
|
data = Controller().model(args)
|
|
data = _truncate_result(data, limit)
|
|
return _xml_response("SSHIntrusionList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("SSHIntrusionList", "error", str(e))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Firewall
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="Firewall", name_cn="List Port Rules", risk_level="low")
|
|
def FirewallPortRuleList(query: str = "", p: int = 1, limit: int = 20) -> str:
|
|
"""
|
|
List port rules with pagination. Each rule has id, ports, protocol, address, types (accept/drop), brief.
|
|
|
|
Args:
|
|
query: Search keyword for port/brief/address. Optional.
|
|
p: Page number. Default 1.
|
|
limit: Items per page. Default 20.
|
|
|
|
Returns: Paginated list of port rules.
|
|
|
|
Workflow: FirewallPortRuleList -> FirewallPortRuleDelete.
|
|
"""
|
|
try:
|
|
from class_v2.safeModelV2.firewallModel import main as FirewallMain
|
|
data = FirewallMain().get_rules_list(_to_obj({"query": query, "p": p, "limit": limit}))
|
|
return _xml_response("FirewallPortRuleList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FirewallPortRuleList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Firewall", name_cn="List IP Rules", risk_level="low")
|
|
def FirewallIPRuleList(limit: int = 200) -> str:
|
|
"""
|
|
List IP rules. Each rule has id, address (IP/CIDR), types (accept/drop), brief.
|
|
|
|
Returns: JSON array of IP rules.
|
|
|
|
Workflow: FirewallIPRuleList -> FirewallIPRuleDelete.
|
|
|
|
Args:
|
|
limit: Max rules to return (default 200). Rules beyond are omitted.
|
|
"""
|
|
try:
|
|
from class_v2.safeModelV2.firewallModel import main as FirewallMain
|
|
data = FirewallMain().get_ip_rules_list(_to_obj({}))
|
|
data = _truncate_result(data, limit)
|
|
return _xml_response("FirewallIPRuleList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FirewallIPRuleList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Firewall", name_cn="List Port Forward Rules", risk_level="low")
|
|
def FirewallForwardList(query: str = "", p: int = 1, limit: int = 20) -> str:
|
|
"""
|
|
(read-only) List port forwarding rules. Each rule has id, start_port, ended_ip, ended_port, protocol.
|
|
|
|
Args:
|
|
query: Search keyword for port. Optional.
|
|
p: Page number. Default 1.
|
|
limit: Items per page. Default 20.
|
|
|
|
Returns: Paginated list of port forwarding rules.
|
|
|
|
Read-only: no tool exists to create/delete forwarding rules. Manage them via the aaPanel Web GUI (Security -> Firewall).
|
|
"""
|
|
try:
|
|
from class_v2.safeModelV2.firewallModel import main as FirewallMain
|
|
data = FirewallMain().get_forward_list(_to_obj({"query": query, "p": p, "limit": limit}))
|
|
return _xml_response("FirewallForwardList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FirewallForwardList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Firewall", name_cn="List Country Rules", risk_level="low")
|
|
def FirewallCountryList(query: str = "", p: int = 1, limit: int = 20) -> str:
|
|
"""
|
|
(read-only) List country/region blocking rules. Each rule has id, country, brief.
|
|
|
|
Args:
|
|
query: Search keyword for country name. Optional.
|
|
p: Page number. Default 1.
|
|
limit: Items per page. Default 20.
|
|
|
|
Returns: Paginated list of country rules.
|
|
|
|
Read-only: no tool exists to create/delete country rules. Manage them via the aaPanel Web GUI (Security -> Firewall).
|
|
"""
|
|
try:
|
|
from class_v2.safeModelV2.firewallModel import main as FirewallMain
|
|
data = FirewallMain().get_country_list(_to_obj({"query": query, "p": p, "limit": limit}))
|
|
return _xml_response("FirewallCountryList", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FirewallCountryList", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Firewall", name_cn="Create Port Rule", risk_level="high")
|
|
def FirewallPortRuleCreate(ports: str, protocol: str, types: str, address: str, brief: str = "") -> str:
|
|
"""
|
|
Create a port firewall rule (accept or drop) for a single source IP.
|
|
|
|
Prefer this over `RunCommand iptables/firewall-cmd` — it persists rules via the panel firewall manager (survives reboots and panel sync). Confirm independent SSH/console access before any firewall change (anti-lockout).
|
|
|
|
Args:
|
|
ports: Port number. Required. e.g. "80" or "80,443".
|
|
protocol: "tcp" or "udp". Required.
|
|
types: "accept" or "drop". Required.
|
|
address: Source IPv4. Required. Must be a single IP (e.g. "192.168.1.100"). All-IPs ("0.0.0.0/0") and CIDR ranges are not supported.
|
|
brief: Remark. Optional.
|
|
|
|
Workflow: FirewallPortRuleList -> FirewallPortRuleCreate.
|
|
"""
|
|
try:
|
|
from class_v2.safeModelV2.firewallModel import main as FirewallMain
|
|
params = {"ports": ports, "protocol": protocol, "types": types,
|
|
"address": address, "source": address, "brief": brief}
|
|
data = FirewallMain().create_rules(_to_obj(params))
|
|
return _xml_response("FirewallPortRuleCreate", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FirewallPortRuleCreate", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Firewall", name_cn="Delete Port Rule", risk_level="high")
|
|
def FirewallPortRuleDelete(id: int, ports: str, address: str, types: str, protocol: str) -> str:
|
|
"""
|
|
Delete a port firewall rule. Irreversible.
|
|
|
|
Args:
|
|
id: Rule ID from FirewallPortRuleList. Required.
|
|
ports: Port number. Required.
|
|
address: Source IP. Required.
|
|
types: "accept" or "drop". Required.
|
|
protocol: "tcp" or "udp". Required.
|
|
|
|
Workflow: FirewallPortRuleList -> FirewallPortRuleDelete.
|
|
"""
|
|
try:
|
|
from class_v2.safeModelV2.firewallModel import main as FirewallMain
|
|
params = {"id": id, "ports": ports, "address": address, "types": types, "protocol": protocol}
|
|
data = FirewallMain().remove_rules(_to_obj(params))
|
|
return _xml_response("FirewallPortRuleDelete", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FirewallPortRuleDelete", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Firewall", name_cn="Create IP Rule", risk_level="high")
|
|
def FirewallIPRuleCreate(address: str, types: str, protocol: str = "tcp", brief: str = "") -> str:
|
|
"""
|
|
Create an IP firewall rule (accept or drop) for a single IP across all ports.
|
|
|
|
Prefer this over `RunCommand iptables/firewall-cmd` — it persists rules via the panel firewall manager. Confirm independent SSH/console access before any firewall change (anti-lockout).
|
|
|
|
Args:
|
|
address: Source IP. Required. Must be a single IP (e.g. "192.168.1.100"). CIDR ranges are not supported.
|
|
types: "accept" or "drop". Required.
|
|
protocol: "tcp" or "udp". Default "tcp".
|
|
brief: Remark. Optional.
|
|
|
|
Workflow: FirewallIPRuleList -> FirewallIPRuleCreate.
|
|
"""
|
|
try:
|
|
from class_v2.safeModelV2.firewallModel import main as FirewallMain
|
|
params = {"address": address, "source": address, "types": types,
|
|
"protocol": protocol, "brief": brief}
|
|
data = FirewallMain().create_ip_rules(_to_obj(params))
|
|
return _xml_response("FirewallIPRuleCreate", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FirewallIPRuleCreate", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Firewall", name_cn="Delete IP Rule", risk_level="high")
|
|
def FirewallIPRuleDelete(id: int, address: str, types: str) -> str:
|
|
"""
|
|
Delete an IP firewall rule. Irreversible.
|
|
|
|
Args:
|
|
id: Rule ID from FirewallIPRuleList. Required.
|
|
address: IP address. Required.
|
|
types: "accept" or "drop". Required.
|
|
|
|
Workflow: FirewallIPRuleList -> FirewallIPRuleDelete.
|
|
"""
|
|
try:
|
|
from class_v2.safeModelV2.firewallModel import main as FirewallMain
|
|
params = {"id": id, "address": address, "types": types}
|
|
data = FirewallMain().remove_ip_rules(_to_obj(params))
|
|
return _xml_response("FirewallIPRuleDelete", "done", json.dumps(data, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("FirewallIPRuleDelete", "error", str(e))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Notification (mod/base/push_mod)
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="Notification", name_cn="List Push Senders", risk_level="low")
|
|
def ListPushSenders() -> str:
|
|
"""
|
|
List configured push channels (notification targets): id, sender_type, used, title.
|
|
|
|
Returns: JSON array. Use sender_type in Notify `channels` to target specific channels.
|
|
|
|
Workflow: ListPushSenders -> Notify.
|
|
"""
|
|
try:
|
|
from mod.base.push_mod.mods import SenderConfig
|
|
senders = [{
|
|
"id": s.get("id"),
|
|
"sender_type": s.get("sender_type"),
|
|
"used": s.get("used"),
|
|
"title": (s.get("data") or {}).get("title", s.get("sender_type")),
|
|
} for s in SenderConfig().config]
|
|
senders = _truncate_result(senders, 100)
|
|
return _xml_response("ListPushSenders", "done", json.dumps(senders, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("ListPushSenders", "error", str(e))
|
|
|
|
|
|
@register_tool(category="Notification", name_cn="Send Notification", risk_level="high")
|
|
def Notify(title: str, message: str, channels: List[str] = None) -> str:
|
|
"""
|
|
Send a notification to configured push channels. HIGH RISK - sends to EXTERNAL services
|
|
(mail/dingding/feishu/webhook/weixin/tg/discord); ensure the message has no secrets.
|
|
|
|
Returns: JSON {sent, total, results:[{channel, ok, error}]}.
|
|
|
|
Workflow: ListPushSenders -> Notify.
|
|
|
|
Args:
|
|
title: Notification title. Required.
|
|
message: Notification body. Required.
|
|
channels: Optional sender_type list, e.g. ["mail","discord"]. Empty = all enabled (sms skipped).
|
|
|
|
Note: After delivering a summary, analysis, or large result, proactively offer to send it via
|
|
this tool (e.g. "want this conclusion sent to your mail/feishu/discord?").
|
|
"""
|
|
try:
|
|
from script.notify_cli import send_notify # shared with the standalone CLI (single source of truth)
|
|
results = send_notify(title, message, channels)
|
|
summary = {"sent": sum(1 for r in results if r["ok"]),
|
|
"total": len(results), "results": results}
|
|
return _xml_response("Notify", "done", json.dumps(summary, ensure_ascii=False))
|
|
except Exception as e:
|
|
return _xml_response("Notify", "error", str(e))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# Security Baseline (read-only scan)
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
@register_tool(category="System", name_cn="Security Baseline Scan", risk_level="low")
|
|
def SecurityBaselineScan(components: List[str] = None) -> str:
|
|
"""
|
|
(read-only) Scan the security baseline; returns FAILED checks per component. Changes nothing.
|
|
|
|
Args:
|
|
components: Subset to scan. Default ["ssh","panel","php","mysql"]. Available: ssh, panel, php, mysql, redis, memcache.
|
|
|
|
Workflow: SecurityBaselineScan -> (report findings; repair manually via panel).
|
|
"""
|
|
try:
|
|
from class_v2.san_baseline_v2 import san_baseline
|
|
sb = san_baseline()
|
|
mapping = {
|
|
"ssh": "ssh_security", "panel": "panel_security", "php": "php_security",
|
|
"mysql": "mysql_security", "redis": "redis_security", "memcache": "memcache_security",
|
|
}
|
|
targets = components if components else ["ssh", "panel", "php", "mysql"]
|
|
results = {}
|
|
for c in targets:
|
|
method = mapping.get(c)
|
|
if not method:
|
|
results[c] = [{"error": "unknown component '{}'".format(c)}]
|
|
continue
|
|
try:
|
|
results[c] = getattr(sb, method)()
|
|
except Exception as e:
|
|
results[c] = [{"error": str(e)}]
|
|
return _xml_response("SecurityBaselineScan", "done",
|
|
json.dumps(results, ensure_ascii=False, default=str))
|
|
except Exception as e:
|
|
return _xml_response("SecurityBaselineScan", "error", str(e))
|