mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-17 21:25:47 +02:00
Jack
a24f2e23c6
1. Added Mail Marketing Automation trigger tasks 2. Added Mail Marketing Groups Import, Export, Merge 3. Added Mail Marketing Subscribers to support paste import 4. Added Mail Marketing Template Import Export Duplicate 5. Added Mail server add timed automatic reply 6. Added HTTPS Protection function for Website (disables automatic HTTP to HTTPS redirection when enabled) 7. Optimize Website Interface Button Integration 8. Optimize the response speed of WP Toolkit interface
544 lines
20 KiB
Python
544 lines
20 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import time
|
|
|
|
from .install import installed, install_git, version_1_5_3
|
|
from .util import read_file, write_file, ExecShell, set_ownership
|
|
from typing import Optional, Dict, Union, List
|
|
from urllib3.util import parse_url, Url
|
|
from mod.base import json_response
|
|
|
|
GIT_TMP_PATH = "/www/server/git_tmp"
|
|
|
|
|
|
class GitTool:
|
|
|
|
def __init__(self, project_path: str, git_url: str, user_config: Optional[dict] = None, git_id: str = ""):
|
|
self.git_url = git_url
|
|
self.get_id = git_id
|
|
self.project_path = project_path
|
|
if self.project_path[-1] == '/':
|
|
self.project_path = self.project_path[:-1]
|
|
if not os.path.isdir(GIT_TMP_PATH):
|
|
os.makedirs(GIT_TMP_PATH)
|
|
|
|
self._tmp_path: Optional[str] = None
|
|
self._askpass_path: Optional[str] = None
|
|
|
|
self.user_config = user_config
|
|
_conf = self.get_user_config_by_project_path()
|
|
if _conf and not self.user_config:
|
|
self.user_config = _conf
|
|
|
|
self._init_tmp_path = False
|
|
|
|
@property
|
|
def tmp_path(self) -> Optional[str]:
|
|
if self._tmp_path is not None:
|
|
return self._tmp_path
|
|
if not os.path.isdir(self.project_path):
|
|
return None
|
|
ino = os.stat(self.project_path).st_ino
|
|
self._tmp_path = "{}/{}".format(GIT_TMP_PATH, str(ino))
|
|
self._askpass_path = "{}/help_{}.sh".format(GIT_TMP_PATH, str(ino))
|
|
return self._tmp_path
|
|
|
|
def get_user_config_by_project_path(self) -> Optional[dict]:
|
|
if os.path.isfile(self.project_path + "/.git/config"):
|
|
git_config = read_file(self.project_path + "/.git/config")
|
|
if isinstance(git_config, str):
|
|
return self._read_user_conf_by_config(git_config)
|
|
return None
|
|
|
|
@property
|
|
def askpass_path(self) -> Optional[str]:
|
|
if self._askpass_path is not None:
|
|
return self._askpass_path
|
|
if not os.path.isdir(self.project_path):
|
|
return None
|
|
ino = os.stat(self.project_path).st_ino
|
|
self._tmp_path = "{}/{}".format(GIT_TMP_PATH, str(ino))
|
|
self._askpass_path = "{}/help_{}.sh".format(GIT_TMP_PATH, str(ino))
|
|
return self._askpass_path
|
|
|
|
def _setup_tmp_path(self) -> Optional[str]:
|
|
if not os.path.isdir(self.project_path):
|
|
return "目标目录：【{}】不存在，无法进行git操作".format(self.project_path)
|
|
|
|
if not os.path.exists(self.tmp_path):
|
|
os.makedirs(self.tmp_path)
|
|
self._init_tmp_path = True
|
|
else:
|
|
if not self._init_tmp_path:
|
|
shutil.rmtree(self.tmp_path)
|
|
os.makedirs(self.tmp_path)
|
|
write_file(self.askpass_path, "")
|
|
ExecShell("chmod +x " + self.askpass_path)
|
|
|
|
if os.path.isdir(self.tmp_path + "/.git"):
|
|
return None
|
|
sh_str = """cd {tmp_path}
|
|
{git_bin} init .
|
|
{git_bin} remote add origin {git_url}
|
|
""".format(tmp_path=self.tmp_path, git_bin=self.git_bin(), git_url=self.git_url)
|
|
|
|
ExecShell(sh_str)
|
|
if not os.path.isfile(self.tmp_path + "/.git/config"):
|
|
return "git Initialization failed"
|
|
|
|
git_conf = read_file(self.tmp_path + "/.git/config")
|
|
if not (isinstance(git_conf, str) and git_conf.find("origin") != -1 and git_conf.find(self.git_url) != -1):
|
|
return "git Failed to set up a remote route"
|
|
|
|
if self.git_url.find("ssh://") != -1: # ssh的情况下不做处理
|
|
return
|
|
|
|
if isinstance(self.user_config, dict):
|
|
sh_str_list = ["cd {}".format(self.tmp_path)]
|
|
config_sh = self.git_bin() + " config user.{} {}"
|
|
for k, v in self.user_config.items():
|
|
if isinstance(k, str) and isinstance(v, str) and k.strip() and v.strip():
|
|
sh_str_list.append(config_sh.format(k, v))
|
|
ExecShell("\n".join(sh_str_list))
|
|
askpass_str = """#!/bin/sh
|
|
case "$1" in
|
|
Username*) exec echo "{}" ;;
|
|
Password*) exec echo "{}" ;;
|
|
esac
|
|
""".format(self.user_config.get('name', "--"), self.user_config.get('password', "--"))
|
|
write_file(self.askpass_path, askpass_str)
|
|
|
|
def remote_branch(self) -> Union[str, List[str]]:
|
|
error = self._setup_tmp_path()
|
|
if error:
|
|
return error
|
|
out, err = ExecShell("cd {} && export GIT_ASKPASS='{}' && git ls-remote origin".format(
|
|
self.tmp_path, self.askpass_path))
|
|
rep_branch = re.compile(r"refs/heads/(?P<b>[^\n]*)\n")
|
|
branch_list = []
|
|
for tmp_res in rep_branch.finditer(out):
|
|
branch_list.append(tmp_res.group("b"))
|
|
|
|
if not branch_list:
|
|
return err
|
|
return branch_list
|
|
|
|
@staticmethod
|
|
def _read_user_conf_by_config(git_config_data: str) -> Optional[dict]:
|
|
rep_user = re.compile(r"\[user][^\n]*\n(?P<target>(\s*\w+\s*=\s*[^\n]*\n)*(\s*\w+\s*=\s*[^\n]*)?)(\s*\[)?")
|
|
res = rep_user.search(git_config_data)
|
|
if not res:
|
|
return None
|
|
res_data = dict()
|
|
k_v_str = res.group("target")
|
|
for line in k_v_str.split("\n"):
|
|
if not line.strip():
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
res_data[k.strip()] = v.strip()
|
|
return res_data
|
|
|
|
@classmethod
|
|
def global_user_conf(cls) -> dict:
|
|
res_dict = {
|
|
"name": None,
|
|
"password": None,
|
|
"email": None,
|
|
}
|
|
global_file = "/root/.gitconfig"
|
|
if not os.path.isfile(global_file):
|
|
return res_dict
|
|
data = read_file(global_file)
|
|
if not isinstance(data, str):
|
|
return res_dict
|
|
res_data = cls._read_user_conf_by_config(data)
|
|
res_dict.update(res_data)
|
|
return res_dict
|
|
|
|
@classmethod
|
|
def set_global_user_conf(cls, data) -> None:
|
|
sh_str = cls.git_bin() + " config --global user.{} {}"
|
|
for k, v in data.items():
|
|
if isinstance(k, str) and isinstance(v, str) and k.strip() and v.strip():
|
|
ExecShell(sh_str.format(k, v))
|
|
|
|
@classmethod
|
|
def ssh_pub_key(cls):
|
|
key_files = ('id_ed25519', 'id_rsa', 'id_ecdsa', 'id_rsa_bt')
|
|
for key_file in key_files:
|
|
key_file = "/root/.ssh/{}".format(key_file)
|
|
pub_file = "/root/.ssh/{}.pub".format(key_file)
|
|
if os.path.isfile(pub_file) and os.path.isfile(key_file):
|
|
data = read_file(pub_file)
|
|
if isinstance(data, str):
|
|
return data
|
|
return cls._create_ssh_key()
|
|
|
|
@staticmethod
|
|
def _create_ssh_key() -> str:
|
|
key_type = "ed25519"
|
|
ExecShell("ssh-keygen -t {s_type} -P '' -f /root/.ssh/id_{s_type} |echo y".format(s_type=key_type))
|
|
authorized_keys = '/root/.ssh/authorized_keys'
|
|
pub_file = "/root/.ssh/id_{s_type}.pub".format(s_type=key_type)
|
|
ExecShell('cat %s >> %s && chmod 600 %s' % (pub_file, authorized_keys, authorized_keys))
|
|
key_type_file = '/www/server/panel/data/ssh_key_type.pl'
|
|
write_file(key_type_file, key_type)
|
|
return read_file(pub_file)
|
|
|
|
@staticmethod
|
|
def git_bin() -> str:
|
|
if not installed():
|
|
if not install_git():
|
|
raise ValueError("There is no git tool and the installation fails, so this feature cannot be used")
|
|
default = "/usr/bin/git"
|
|
git_path = shutil.which("git")
|
|
if git_path is None:
|
|
return default
|
|
return git_path
|
|
|
|
def pull(self, branch, set_own: Optional[str] = None) -> Optional[str]:
|
|
if self.git_url.startswith("https://") or self.git_url.startswith("http://"):
|
|
res = parse_url(self.git_url)
|
|
if isinstance(res, Url) and not res.auth:
|
|
if self.user_config and "name" in self.user_config and "password" in self.user_config:
|
|
res.auth = "{}:{}".format(self.user_config["name"], self.user_config["password"])
|
|
self.git_url = res.url
|
|
git_name = self.git_name()
|
|
if git_name is None:
|
|
git_name = 'None'
|
|
if os.path.isdir(self.tmp_path + "/" + git_name):
|
|
shutil.rmtree(self.tmp_path + "/" + git_name)
|
|
|
|
log_file = "/tmp/git_{}_log.log".format(self.get_id)
|
|
|
|
shell_command_str = "cd {0} && {1} clone --progress -b {2} {3} &>> {4}".format(
|
|
self.tmp_path, self.git_bin(), branch, self.git_url, log_file
|
|
)
|
|
|
|
ExecShell(shell_command_str)
|
|
if not os.path.isdir(self.tmp_path + "/" + git_name):
|
|
return "Pull error"
|
|
|
|
ExecShell(r"\cp -rf {}/{}/* {}/".format(self.tmp_path, git_name, self.project_path))
|
|
if isinstance(set_own, str):
|
|
set_ownership(self.project_path, set_own)
|
|
|
|
if os.path.isdir(self.tmp_path + "/" + git_name):
|
|
shutil.rmtree(self.tmp_path + "/" + git_name)
|
|
|
|
def git_name(self) -> Optional[str]:
|
|
if isinstance(self.git_url, str):
|
|
name = self.git_url.rsplit("/", 1)[1]
|
|
if name.endswith(".git"):
|
|
name = name[:-4]
|
|
return name
|
|
return None
|
|
|
|
@classmethod
|
|
def new_id(cls) -> str:
|
|
from uuid import uuid4
|
|
return uuid4().hex[::2]
|
|
|
|
|
|
class RealGitMager:
|
|
_git_config_file = "/www/server/panel/data/site_git_config.json"
|
|
|
|
def __init__(self):
|
|
self._config: Optional[Dict[str, List[Dict[str, Union[int, str, dict]]]]] = None
|
|
# c = {
|
|
# "site_name": [{
|
|
# "id": "",
|
|
# "site_name": "aaaa",
|
|
# "url": "http://git.bt.cn/cjxin/panel-plugin.git", # ssh://git@git.bt.cn/cjxin/panel-plugin.git
|
|
# "path_ino": 4564524,
|
|
# "git_path": "/www/wwwroot/site",
|
|
# "config": {
|
|
# "name": "",
|
|
# "password": "",
|
|
# "email": "",
|
|
# },
|
|
# }
|
|
# ]
|
|
# }
|
|
|
|
@property
|
|
def configure(self) -> Dict[str, List[Dict[str, Union[int, str, dict]]]]:
|
|
if self._config is None:
|
|
try:
|
|
res = read_file(self._git_config_file)
|
|
if res is None:
|
|
data = {}
|
|
else:
|
|
data = json.loads(res)
|
|
except (json.JSONDecoder, TypeError, ValueError):
|
|
data = {}
|
|
|
|
self._config = data
|
|
return self._config
|
|
|
|
def save_configure(self):
|
|
if self._config:
|
|
write_file(self._git_config_file, json.dumps(self._config))
|
|
|
|
def add_git(self, git_url: str, site_name: str, git_path: str, user_config: Optional[dict]) -> Union[str, list]:
|
|
url = parse_url(git_url)
|
|
if not (isinstance(url, Url) and url.scheme and url.host and url.path):
|
|
return "The URL format is incorrect"
|
|
|
|
if user_config and not (isinstance(user_config, dict) and "name" in user_config and "password" in user_config):
|
|
return "User information is entered incorrectly"
|
|
|
|
if not os.path.exists(git_path):
|
|
return "The git target directory does not exist"
|
|
else:
|
|
path_ino = os.stat(git_path).st_ino
|
|
|
|
if site_name not in self.configure:
|
|
self.configure[site_name] = []
|
|
|
|
for c in self.configure[site_name]:
|
|
if c["path_ino"] == path_ino or git_path == c["git_path"]:
|
|
return "The path already exists, so please do not add it repeatedly"
|
|
|
|
try:
|
|
GitTool.git_bin()
|
|
except ValueError as e:
|
|
return str(e)
|
|
|
|
git_id = GitTool.new_id()
|
|
git = GitTool(project_path=git_path, git_url=git_url, user_config=user_config, git_id=git_id)
|
|
res = git.remote_branch()
|
|
if isinstance(res, str):
|
|
return res
|
|
|
|
self.configure[site_name].append(
|
|
{
|
|
"id": git_id,
|
|
"site_name": site_name,
|
|
"url": git_url,
|
|
"path_ino": path_ino,
|
|
"git_path": git_path,
|
|
"remote_branch": res,
|
|
"remote_branch_time": int(time.time()),
|
|
"config": user_config,
|
|
}
|
|
)
|
|
self.save_configure()
|
|
return res
|
|
|
|
def modify_git(
|
|
self,
|
|
git_id: str,
|
|
site_name: str,
|
|
git_url: Optional[str],
|
|
git_path: Optional[str],
|
|
user_config: Optional[dict]
|
|
) -> Optional[str]:
|
|
target = None
|
|
for i in self.configure.get(site_name, []):
|
|
if i["id"] == git_id:
|
|
target = i
|
|
break
|
|
|
|
if target is None:
|
|
return 'The specified git configuration does not exist'
|
|
if git_url:
|
|
url = parse_url(git_url)
|
|
if not (isinstance(url, Url) and url.scheme and url.host and url.path):
|
|
return "The URL format is incorrect"
|
|
target["url"] = git_url
|
|
|
|
if git_path:
|
|
if not os.path.exists(git_path):
|
|
return "The git target directory does not exist"
|
|
else:
|
|
path_ino = os.stat(git_path).st_ino
|
|
target["path_ino"] = path_ino
|
|
target["git_path"] = git_path
|
|
|
|
if user_config:
|
|
if not (isinstance(user_config, dict) and "name" in user_config and "password" in user_config):
|
|
return "User information is entered incorrectly"
|
|
target["config"] = user_config
|
|
|
|
git = GitTool(project_path=target['git_path'],
|
|
git_url=target['url'],
|
|
user_config=target['config'],
|
|
git_id=target['id'])
|
|
res = git.remote_branch()
|
|
if isinstance(res, str):
|
|
return res
|
|
|
|
self.save_configure()
|
|
return None
|
|
|
|
def remove_git(self, git_id: str, site_name: str) -> Optional[str]:
|
|
target = None
|
|
for i in self.configure.get(site_name, []):
|
|
if i["id"] == git_id:
|
|
target = i
|
|
break
|
|
|
|
if target is None:
|
|
return 'The specified git configuration does not exist'
|
|
|
|
self.configure[site_name].remove(target)
|
|
self.save_configure()
|
|
|
|
def site_git_configure(self, site_name, refresh: bool = False) -> List[dict]:
|
|
if site_name not in self.configure:
|
|
return []
|
|
res_list = []
|
|
for i in self.configure[site_name]:
|
|
if time.time() - i.get("remote_branch_time", 0) > 60 * 60 or refresh:
|
|
g = GitTool(project_path=i['git_path'], git_url=i['url'], user_config=i['config'], git_id=i['id'])
|
|
res = g.remote_branch()
|
|
if isinstance(res, str):
|
|
i.update(remote_branch_error=res, remote_branch=[], remote_branch_time=int(time.time()))
|
|
else:
|
|
i.update(remote_branch=res, remote_branch_time=int(time.time()))
|
|
res_list.append(i)
|
|
|
|
return res_list
|
|
|
|
@staticmethod
|
|
def set_global_user(name: Optional[str], password: Optional[str], email: Optional[str] = None) -> None:
|
|
data = {}
|
|
if name:
|
|
data['name'] = name
|
|
if password:
|
|
data['password'] = password
|
|
if email:
|
|
data['email'] = email
|
|
GitTool.set_global_user_conf(data)
|
|
|
|
def git_pull(self, git_id: str, site_name: str, branch: str) -> Optional[str]:
|
|
target = None
|
|
for i in self.configure.get(site_name, []):
|
|
if i["id"] == git_id:
|
|
target = i
|
|
break
|
|
|
|
if target is None:
|
|
return 'The specified git configuration does not exist'
|
|
|
|
g = GitTool(
|
|
project_path=target['git_path'],
|
|
git_url=target['url'],
|
|
user_config=target['config'],
|
|
git_id=target["id"])
|
|
return g.pull(branch)
|
|
|
|
|
|
class GitMager:
|
|
# 添加git信息
|
|
@staticmethod
|
|
def add_git(get):
|
|
user_config = None
|
|
try:
|
|
git_url = get.url.strip()
|
|
site_name = get.site_name.strip()
|
|
git_path = get.git_path.strip()
|
|
if hasattr(get, "config") and get.config.strip():
|
|
user_config = json.loads(get.config.strip())
|
|
except (json.JSONDecoder, AttributeError, TypeError):
|
|
return json_response(status=False, msg="The parameter is incorrect")
|
|
res = RealGitMager().add_git(git_url, site_name, git_path, user_config)
|
|
if isinstance(res, str):
|
|
return json_response(status=False, msg=res)
|
|
return json_response(status=True, data=res)
|
|
|
|
# 修改git信息
|
|
@staticmethod
|
|
def modify_git(get):
|
|
git_url = None
|
|
git_path = None
|
|
user_config = None
|
|
try:
|
|
git_id = get.git_id.strip()
|
|
site_name = get.site_name.strip()
|
|
if "url" in get:
|
|
git_url = get.url.strip()
|
|
if 'git_path' in get:
|
|
git_path = get.git_path.strip()
|
|
if hasattr(get, "user_config") and get.user_config.strip():
|
|
user_config = json.loads(get.user_config.strip())
|
|
except (json.JSONDecoder, AttributeError, TypeError):
|
|
return json_response(status=False, msg="parameter error")
|
|
res = RealGitMager().modify_git(git_id, site_name, git_url, git_path, user_config)
|
|
if isinstance(res, str):
|
|
return json_response(status=False, msg=res)
|
|
return json_response(status=True, data=res)
|
|
|
|
# 移除git信息
|
|
@staticmethod
|
|
def remove_git(get):
|
|
try:
|
|
git_id = get.git_id.strip()
|
|
site_name = get.site_name.strip()
|
|
except (json.JSONDecoder, AttributeError, TypeError):
|
|
return json_response(status=False, msg="parameter error")
|
|
res = RealGitMager().remove_git(git_id, site_name)
|
|
if isinstance(res, str):
|
|
return json_response(status=False, msg=res)
|
|
return json_response(status=True, data=res)
|
|
|
|
@staticmethod
|
|
def site_git_configure(get):
|
|
if not version_1_5_3():
|
|
return json_response(status=False, msg="Git versions earlier than 1.5.3 are not available")
|
|
refresh = ''
|
|
try:
|
|
site_name = get.site_name.strip()
|
|
if "refresh" in get:
|
|
refresh = get.refresh.strip()
|
|
except (AttributeError, TypeError):
|
|
return json_response(status=False, msg="parameter error")
|
|
if refresh in ("true", "1"):
|
|
refresh = True
|
|
else:
|
|
refresh = False
|
|
res = RealGitMager().site_git_configure(site_name, refresh=refresh)
|
|
if isinstance(res, str):
|
|
return json_response(status=False, msg=res)
|
|
return json_response(status=True, data=res)
|
|
|
|
@staticmethod
|
|
def set_global_user(get):
|
|
name = password = email = None
|
|
try:
|
|
if "name" in get:
|
|
name = get.name.strip()
|
|
if "password" in get:
|
|
password = get.password.strip()
|
|
if "email" in get:
|
|
email = get.email.strip()
|
|
except (AttributeError, TypeError):
|
|
return json_response(status=False, msg="parameter error")
|
|
|
|
RealGitMager().set_global_user(name, password, email)
|
|
return json_response(status=True, msg="The setup was successful")
|
|
|
|
@staticmethod
|
|
def git_pull(get):
|
|
try:
|
|
site_name = get.site_name.strip()
|
|
git_id = get.git_id.strip()
|
|
branch = get.branch.strip()
|
|
except (AttributeError, TypeError):
|
|
return json_response(status=False, msg="parameter error")
|
|
res = RealGitMager().git_pull(git_id, site_name, branch)
|
|
if isinstance(res, str):
|
|
return json_response(status=False, msg=res)
|
|
return json_response(status=True, data=res)
|
|
|
|
@staticmethod
|
|
def git_global_user_conf(get=None):
|
|
return GitTool.global_user_conf()
|
|
|
|
@staticmethod
|
|
def git_ssh_pub_key(get=None):
|
|
return GitTool.ssh_pub_key()
|