364 lines
15 KiB
Python
364 lines
15 KiB
Python
import os
|
||
import re
|
||
import json
|
||
from dataclasses import dataclass
|
||
from typing import Tuple, Optional, Union, Dict
|
||
from .util import webserver, check_server_config, DB, \
|
||
write_file, read_file, GET_CLASS, service_reload, pre_re_key
|
||
from mod.base import json_response
|
||
|
||
|
||
@dataclass
|
||
class _RefererConf:
|
||
name: str
|
||
fix: str
|
||
domains: str
|
||
status: str
|
||
return_rule: str
|
||
http_status: str
|
||
|
||
def __str__(self):
|
||
return '{"name"="%s","fix"="%s","domains"="%s","status"="%s","http_status"="%s","return_rule"="%s"}' % (
|
||
self.name, self.fix, self.domains, self.status, self.http_status, self.return_rule
|
||
)
|
||
|
||
|
||
class RealReferer:
|
||
_referer_conf_dir = '/www/server/panel/vhost/config' # 防盗链配置
|
||
_ng_referer_conf_format = r''' #SECURITY-START 防盗链配置
|
||
location ~ .*\.(%s)$ {
|
||
expires 30d;
|
||
access_log /dev/null;
|
||
valid_referers %s;
|
||
if ($invalid_referer){
|
||
%s;
|
||
}
|
||
}
|
||
#SECURITY-END'''
|
||
|
||
def __init__(self, config_prefix: str):
|
||
if not os.path.isdir(self._referer_conf_dir):
|
||
os.makedirs(self._referer_conf_dir)
|
||
self.config_prefix: str = config_prefix
|
||
self._webserver = None
|
||
|
||
@property
|
||
def webserver(self) -> str:
|
||
if self._webserver is not None:
|
||
return self._webserver
|
||
self._webserver = webserver()
|
||
return self._webserver
|
||
|
||
def get_config(self, site_name: str) -> Optional[_RefererConf]:
|
||
try:
|
||
config = json.loads(read_file("{}/{}{}_door_chain.json".format(self._referer_conf_dir, self.config_prefix, site_name)))
|
||
except (json.JSONDecodeError, TypeError, ValueError):
|
||
config = None
|
||
if isinstance(config, dict):
|
||
return _RefererConf(**config)
|
||
return None
|
||
|
||
def save_config(self, site_name: str, data: Union[dict, str, _RefererConf]) -> bool:
|
||
if isinstance(data, dict):
|
||
c = json.dumps(data)
|
||
elif isinstance(data, _RefererConf):
|
||
c = json.dumps(str(data))
|
||
else:
|
||
c = data
|
||
|
||
file_path = "{}/{}{}_door_chain.json".format(self._referer_conf_dir, self.config_prefix, site_name)
|
||
return write_file(file_path, c)
|
||
|
||
# 检测参数，如果正确则返回 配置数据类型的值，否则返回错误信息
|
||
@staticmethod
|
||
def check_args(get: Union[Dict, GET_CLASS]) -> Union[_RefererConf, str]:
|
||
res = {}
|
||
if isinstance(get, GET_CLASS):
|
||
try:
|
||
res["status"] = "true" if not hasattr(get, "status") else get.status.strip()
|
||
res["http_status"] = "false" if not hasattr(get, "http_status") else get.http_status.strip()
|
||
res["name"] = get.name.strip()
|
||
res["fix"] = get.fix.strip()
|
||
res["domains"] = get.domains.strip()
|
||
res["return_rule"] = get.return_rule.strip()
|
||
except AttributeError:
|
||
return "Parameter error"
|
||
else:
|
||
try:
|
||
res["status"] = "true" if "status" not in get else get["status"].strip()
|
||
res["http_status"] = "false" if "http_status" not in get else get["http_status"].strip()
|
||
res["name"] = get["name"].strip()
|
||
res["fix"] = get["fix"].strip()
|
||
res["domains"] = get["domains"].strip()
|
||
res["return_rule"] = get["return_rule"].strip()
|
||
except KeyError:
|
||
return "Parameter error"
|
||
|
||
rconf = _RefererConf(**res)
|
||
if rconf.status not in ("true", "false") and rconf.return_rule not in ("true", "false"):
|
||
return "状态参数只能使用【true,false】"
|
||
if rconf.return_rule not in ('404', '403', '200', '301', '302', '401') and rconf.return_rule[0] != "/":
|
||
return "响应资源应使用URI路径或HTTP状态码，如：/test.png 或 404"
|
||
if len(rconf.domains) < 3:
|
||
return "防盗链域名不能为空"
|
||
if len(rconf.fix) < 2:
|
||
return 'URL后缀不能为空!'
|
||
return rconf
|
||
|
||
def set_referer_security(self, rc: _RefererConf) -> Tuple[bool, str]:
|
||
error_msg = self._set_nginx_referer_security(rc)
|
||
if error_msg and self.webserver == "nginx":
|
||
return False, error_msg
|
||
error_msg = self._set_apache_referer_security(rc)
|
||
if error_msg and self.webserver == "apache":
|
||
return False, error_msg
|
||
service_reload()
|
||
self.save_config(rc.name, rc)
|
||
return True, "设置成功"
|
||
|
||
def _set_nginx_referer_security(self, rc: _RefererConf) -> Optional[str]:
|
||
ng_file = '/www/server/panel/vhost/nginx/{}{}.conf'.format(self.config_prefix, rc.name)
|
||
ng_conf = read_file(ng_file)
|
||
if not isinstance(ng_conf, str):
|
||
return "nginx配置文件丢失，无法设置"
|
||
start_idx, end_idx = self._get_nginx_referer_security_idx(ng_conf)
|
||
if rc.status == "true":
|
||
if rc.return_rule[0] == "/":
|
||
return_rule = "rewrite /.* {} break".format(rc.return_rule)
|
||
else:
|
||
return_rule = 'return {}'.format(rc.return_rule)
|
||
|
||
valid_args_list = []
|
||
if rc.http_status == "true":
|
||
valid_args_list.extend(("none", "blocked"))
|
||
valid_args_list.extend(map(lambda x: x.strip(), rc.domains.split(",")))
|
||
valid_args = " ".join(valid_args_list)
|
||
|
||
location_args = "|".join(map(lambda x: pre_re_key(x.strip()), rc.fix.split(",")))
|
||
if start_idx is not None:
|
||
new_conf = ng_conf[:start_idx] + "\n" + (
|
||
self._ng_referer_conf_format % (location_args, valid_args, return_rule)
|
||
) + "\n" + ng_conf[end_idx:]
|
||
else:
|
||
rep_redirect_include = re.compile(r"\sinclude +.*/redirect/.*\*\.conf;", re.M)
|
||
redirect_include_res = rep_redirect_include.search(ng_conf)
|
||
if redirect_include_res:
|
||
new_conf = ng_conf[:redirect_include_res.end()] + "\n" + (
|
||
self._ng_referer_conf_format % (location_args, valid_args, return_rule)
|
||
) + ng_conf[redirect_include_res.end():]
|
||
else:
|
||
if "#SSL-END" not in ng_conf:
|
||
return "添加配置失败，无法定位SSL相关配置的位置"
|
||
|
||
new_conf = ng_conf.replace("#SSL-END", "#SSL-END\n" + self._ng_referer_conf_format % (
|
||
location_args, valid_args, return_rule))
|
||
|
||
else:
|
||
if start_idx is None:
|
||
return
|
||
new_conf = ng_conf[:start_idx] + "\n" + ng_conf[end_idx:]
|
||
|
||
write_file(ng_file, new_conf)
|
||
if self.webserver == "nginx" and check_server_config() is not None:
|
||
write_file(ng_file, ng_conf)
|
||
return "配置失败"
|
||
|
||
@staticmethod
|
||
def _get_nginx_referer_security_idx(ng_conf: str) -> Tuple[Optional[int], Optional[int]]:
|
||
rep_security = re.compile(
|
||
r"(\s*#\s*SECURITY-START.*\n)?\s*location\s+~\s+\.\*\\\.\(.*(\|.*)?\)\$\s*\{[^}]*valid_referers"
|
||
)
|
||
res = rep_security.search(ng_conf)
|
||
if res is None:
|
||
return None, None
|
||
|
||
start_idx = res.start()
|
||
s_idx = start_idx + ng_conf[start_idx:].find("{") + 1 # 起始位置
|
||
l_n = 1
|
||
max_idx = len(ng_conf)
|
||
while l_n > 0:
|
||
next_l = ng_conf[s_idx:].find("{")
|
||
next_r = ng_conf[s_idx:].find("}") # 可能存在报错
|
||
if next_r == -1:
|
||
return None, None
|
||
if next_l == -1:
|
||
next_l = max_idx
|
||
|
||
if next_l < next_r:
|
||
l_n += 1
|
||
else:
|
||
l_n -= 1
|
||
s_idx += min(next_l, next_r) + 1
|
||
|
||
rep_comment = re.search(r"^\s*#\s*SECURITY-END[^\n]*\n", ng_conf[s_idx:])
|
||
if rep_comment is not None:
|
||
end_idx = s_idx + rep_comment.end()
|
||
else:
|
||
end_idx = s_idx
|
||
|
||
return start_idx, end_idx
|
||
|
||
@staticmethod
|
||
def _build_apache_referer_security_conf(rc: _RefererConf) -> str:
|
||
r_conf_list = ["#SECURITY-START 防盗链配置"]
|
||
cond_format = " RewriteCond %{{HTTP_REFERER}} !{} [NC]"
|
||
if rc.http_status == "false":
|
||
r_conf_list.append(cond_format.format("^$"))
|
||
|
||
r_conf_list.extend(map(lambda x: cond_format.format(x.strip()), rc.domains.split(",")))
|
||
|
||
rule_format = " RewriteRule .({}) {} "
|
||
if rc.return_rule[0] == "/":
|
||
r_conf_list.append(rule_format.format(
|
||
"|".join(map(lambda x: x.strip(), rc.fix.split(","))),
|
||
rc.return_rule
|
||
))
|
||
else:
|
||
r_conf_list.append(rule_format.format(
|
||
"|".join(map(lambda x: x.strip(), rc.fix.split(","))),
|
||
"/{s}.html [R={s},NC,L]".format(s=rc.return_rule)
|
||
))
|
||
|
||
r_conf_list.append(" #SECURITY-END")
|
||
|
||
return "\n".join(r_conf_list)
|
||
|
||
# 根据配置正则确定位置 并将配置文件添加进去 use_start 参数指定添加的前后
|
||
def _add_apache_referer_security_by_rep_idx(self,
|
||
rep: re.Pattern,
|
||
use_start: bool,
|
||
ap_conf, ap_file, r_conf) -> bool:
|
||
tmp_conf_list = []
|
||
last_idx = 0
|
||
for tmp in rep.finditer(ap_conf):
|
||
tmp_conf_list.append(ap_conf[last_idx:tmp.start()])
|
||
if use_start:
|
||
tmp_conf_list.append("\n" + r_conf + "\n")
|
||
tmp_conf_list.append(tmp.group())
|
||
else:
|
||
tmp_conf_list.append(tmp.group())
|
||
tmp_conf_list.append("\n" + r_conf + "\n")
|
||
last_idx = tmp.end()
|
||
if last_idx == 0:
|
||
return False
|
||
|
||
tmp_conf_list.append(ap_conf[last_idx:])
|
||
_conf = "".join(tmp_conf_list)
|
||
write_file(ap_file, _conf)
|
||
if self.webserver == "apache" and check_server_config() is not None:
|
||
write_file(ap_file, ap_conf)
|
||
return False
|
||
return True
|
||
|
||
def _set_apache_referer_security(self, rc: _RefererConf) -> Optional[str]:
|
||
ap_file = '/www/server/panel/vhost/apache/{}{}.conf'.format(self.config_prefix, rc.name)
|
||
ap_conf = read_file(ap_file)
|
||
if not isinstance(ap_conf, str):
|
||
return "nginx配置文件丢失，无法设置"
|
||
rep_security = re.compile(r"#\s*SECURITY-START(.|\n)#SECURITY-END.*\n")
|
||
res = rep_security.search(ap_conf)
|
||
if rc.status == "true":
|
||
r_conf = self._build_apache_referer_security_conf(rc)
|
||
if res is not None:
|
||
new_conf_list = []
|
||
_idx = 0
|
||
for tmp_res in rep_security.finditer(ap_conf):
|
||
new_conf_list.append(ap_conf[_idx:tmp_res.start()])
|
||
new_conf_list.append("\n" + r_conf + "\n")
|
||
_idx = tmp_res.end()
|
||
new_conf_list.append(ap_conf[_idx:])
|
||
new_conf = "".join(new_conf_list)
|
||
write_file(ap_file, new_conf)
|
||
if self.webserver == "apache" and check_server_config() is not None:
|
||
write_file(ap_file, ap_conf)
|
||
return "配置修改失败"
|
||
|
||
rep_redirect_include = re.compile(r"IncludeOptional +.*/redirect/.*\*\.conf.*\n", re.M)
|
||
rep_custom_log = re.compile(r"CustomLog .*\n")
|
||
rep_deny_files = re.compile(r"\n\s*#DENY FILES")
|
||
if self._add_apache_referer_security_by_rep_idx(rep_redirect_include, False, ap_conf, ap_file, r_conf):
|
||
return
|
||
if self._add_apache_referer_security_by_rep_idx(rep_custom_log, False, ap_conf, ap_file, r_conf):
|
||
return
|
||
if self._add_apache_referer_security_by_rep_idx(rep_deny_files, True, ap_conf, ap_file, r_conf):
|
||
return
|
||
return "设置添加失败"
|
||
|
||
else:
|
||
if res is None:
|
||
return
|
||
|
||
new_conf_list = []
|
||
_idx = 0
|
||
for tmp_res in rep_security.finditer(ap_conf):
|
||
new_conf_list.append(ap_conf[_idx:tmp_res.start()])
|
||
_idx = tmp_res.end()
|
||
new_conf_list.append(ap_conf[_idx:])
|
||
new_conf = "".join(new_conf_list)
|
||
write_file(ap_file, new_conf)
|
||
if self.webserver == "apache" and check_server_config() is not None:
|
||
write_file(ap_file, ap_conf)
|
||
return "配置修改失败"
|
||
|
||
def get_referer_security(self, site_name) -> Optional[dict]:
|
||
r = self.get_config(site_name)
|
||
if r is None:
|
||
return None
|
||
return json.loads(str(r))
|
||
|
||
def remove_site_referer_info(self, site_name):
|
||
file_path = "{}/{}{}_door_chain.json".format(self._referer_conf_dir, self.config_prefix, site_name)
|
||
if os.path.exists(file_path):
|
||
os.remove(file_path)
|
||
|
||
# 从配置文件中获取referer配置信息
|
||
# 暂时不实现，意义不大
|
||
def _get_referer_security_by_conf(self, site_name):
|
||
if self.webserver == "nginx":
|
||
self._get_nginx_referer_security()
|
||
else:
|
||
self._get_apache_referer_security()
|
||
|
||
|
||
class Referer:
|
||
|
||
def __init__(self, config_prefix: str):
|
||
self.config_prefix: str = config_prefix
|
||
self._r = RealReferer(self.config_prefix)
|
||
|
||
def get_referer_security(self, get):
|
||
try:
|
||
site_name = get.site_name.strip()
|
||
except AttributeError:
|
||
return json_response(status=False, msg="Parameter error")
|
||
|
||
data = self._r.get_referer_security(site_name)
|
||
if data is None:
|
||
default_conf = {
|
||
"name": site_name,
|
||
"fix": "jpg,jpeg,gif,png,js,css",
|
||
"domains": "",
|
||
"status": "false",
|
||
"return_rule": "404",
|
||
"http_status": "false",
|
||
}
|
||
site_info = DB("sites").where("name=?", (site_name,)).field('id').find()
|
||
if not isinstance(site_info, dict):
|
||
return json_response(status=False, msg="Site query error")
|
||
domains_info = DB("domain").where("pid=?", (site_info["id"],)).field('name').select()
|
||
if not isinstance(domains_info, list):
|
||
return json_response(status=False, msg="Site query error")
|
||
|
||
default_conf["domains"] = ",".join(map(lambda x: x["name"], domains_info))
|
||
return json_response(status=True, data=default_conf)
|
||
|
||
return json_response(status=True, data=data)
|
||
|
||
def set_referer_security(self, get):
|
||
r = self._r.check_args(get)
|
||
if isinstance(r, str):
|
||
return json_response(status=False, msg=r)
|
||
|
||
flag, msg = self._r.set_referer_security(r)
|
||
return json_response(status=flag, msg=msg)
|