fix #41

CHANGELOG.md

@@ -6,6 +6,11 @@
     - ssl证书申请
     - admin拥有所有权限
 
+- v1.4.32(2023-07-10)
+    - 修复 部分代码在Python2版本下兼容性问题
+    - 新增 工具箱目录下新页面：域名子域名查询
+    - 新增 添加域名时可选项：子域证书，可以自动添加该域名下子域名到证书列表 [issues#41](https://github.com/mouday/domain-admin/issues/41)
+
 - v1.4.31(2023-07-09)
     - 修复 只读用户无法使用搜索功能bug
     - 优化 手动编辑更新证书信息接口时间过长的问题

domain_admin/api/domain_api.py

@@ -3,6 +3,7 @@
 由于历史原因，domain指代 SSL证书的域名
 """
 from __future__ import print_function, unicode_literals, absolute_import, division
+
 from operator import itemgetter
 
 from flask import request, g
@@ -611,4 +612,4 @@ def get_domain_group_filter():
     return {
         'list': lst,
         'total': len(lst),
-    }
+    }

domain_admin/api/domain_info_api.py

@@ -16,9 +16,10 @@
 from domain_admin.model.group_model import GroupModel
 from domain_admin.model.group_user_model import GroupUserModel
 from domain_admin.service import domain_info_service, async_task_service, file_service, group_service, \
-    operation_service, group_user_service
+    operation_service, group_user_service, domain_service
 from domain_admin.utils import domain_util, time_util, icp_util
 from domain_admin.utils.flask_ext.app_exception import AppException
+from domain_admin.utils.open_api import crtsh_api
 
 
 @operation_service.operation_log_decorator(
@@ -38,6 +39,7 @@ def add_domain_info():
     domain_start_time = request.json.get('domain_start_time')
     domain_expire_time = request.json.get('domain_expire_time')
     is_auto_update = request.json.get('is_auto_update', True)
+    is_auto_subdomain = request.json.get('is_auto_subdomain', False)
     comment = request.json.get('comment', '')
     group_id = request.json.get('group_id') or 0
 
@@ -51,6 +53,15 @@ def add_domain_info():
         is_auto_update=is_auto_update
     )
 
+    # 异步提交
+    if is_auto_subdomain:
+        async_task_service.submit_task(
+            fn=domain_service.auto_import_from_domain,
+            root_domain=domain,
+            group_id=group_id,
+            user_id=current_user_id
+        )
+
     return {'domain_info_id': row.id}
 
 
@@ -73,6 +84,7 @@ def update_domain_info_by_id():
     domain_start_time = request.json.get('domain_start_time')
     domain_expire_time = request.json.get('domain_expire_time')
     is_auto_update = request.json.get('is_auto_update', True)
+    is_auto_subdomain = request.json.get('is_auto_subdomain', False)
     comment = request.json.get('comment', '')
     group_id = request.json.get('group_id') or 0
 
@@ -101,6 +113,13 @@ def update_domain_info_by_id():
         # 需要自动更新
         domain_info_service.update_domain_info_row(domain_info_row)
 
+    if is_auto_subdomain:
+        async_task_service.submit_task(
+            fn=domain_service.auto_import_from_domain,
+            root_domain=domain,
+            group_id=group_id,
+            user_id=current_user_id
+        )
 
 @operation_service.operation_log_decorator(
     model=DomainInfoModel,
@@ -469,3 +488,18 @@ def get_icp():
     domain = request.json['domain']
     res = icp_util.get_icp(domain)
     return res.get('info')
+
+
+def get_sub_domain_cert():
+    """
+    获取子域证书列表
+    :return:
+    """
+    keyword = request.json.get('keyword', 1)
+
+    lst = crtsh_api.search(keyword)
+
+    return {
+        'list': lst,
+        'total': len(lst)
+    }

domain_admin/model/base_model.py

@@ -1,10 +1,6 @@
 # -*- coding: utf-8 -*-
 from __future__ import print_function, unicode_literals, absolute_import, division
 
-import warnings
-
-warnings.filterwarnings("ignore")
-
 import logging
 from logging.handlers import RotatingFileHandler
 

domain_admin/router/api_map.py

@@ -129,6 +129,7 @@
     '/api/importDomainInFromFile': domain_info_api.import_domain_info_from_file,
     '/api/exportDomainInfoFile': domain_info_api.export_domain_info_file,
     '/api/getDomainInfoGroupFilter': domain_info_api.get_domain_info_group_filter,
+    '/api/getSubDomainCert': domain_info_api.get_sub_domain_cert,
 
     # prometheus
     '/metrics': prometheus_api.metrics,

domain_admin/service/domain_service.py

@@ -22,6 +22,7 @@
 from domain_admin.utils import domain_util
 from domain_admin.utils.cert_util import cert_socket_v2, cert_openssl_v2
 from domain_admin.utils.flask_ext.app_exception import ForbiddenAppException
+from domain_admin.utils.open_api import crtsh_api
 
 
 def update_domain_host_list(domain_row):
@@ -311,6 +312,41 @@ def check_permission_and_get_row(domain_id, user_id):
     return row
 
 
+def auto_import_from_domain(root_domain, group_id=0, user_id=0):
+    """
+    自动导入顶级域名下包含的子域名到证书列表
+    :param root_domain: str
+    :param group_id: int
+    :param user_id: int
+    :return:
+    """
+    lst = crtsh_api.search(root_domain)
+
+    domain_set = list(set([domain['common_name'] for domain in lst]))
+
+    data = [
+        {
+            'domain': domain,
+            'root_domain': root_domain,
+            'port': 443,
+            'alias': '',
+            'user_id': user_id,
+            'group_id': group_id,
+        } for domain in domain_set
+    ]
+
+    for batch in chunked(data, 500):
+        DomainModel.insert_many(batch).on_conflict_ignore().execute()
+
+    # 更新插入的证书
+    rows = DomainModel.select().where(
+        DomainModel.domain.in_(domain_set)
+    )
+
+    for row in rows:
+        update_domain_row(row)
+
+
 def add_domain_from_file(filename, user_id):
     logger.info('user_id: %s, filename: %s', user_id, filename)
 

domain_admin/utils/cert_util/cert_openssl_v2.py

@@ -82,8 +82,10 @@ def get_ssl_cert_by_openssl(
     ssl_context.verify_mode = ssl.CERT_NONE
     ssl_context.check_hostname = False
 
-    with ssl_context.wrap_socket(sock, server_hostname=domain) as wrap_socket:
-        dercert = wrap_socket.getpeercert(True)
+    # fix: Python2 AttributeError: __exit__
+    wrap_socket = ssl_context.wrap_socket(sock, server_hostname=domain)
+    dercert = wrap_socket.getpeercert(True)
+    wrap_socket.close()
 
     server_cert = ssl.DER_cert_to_PEM_cert(dercert)
     cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, server_cert.encode())

domain_admin/utils/cert_util/cert_socket_v2.py

@@ -57,8 +57,12 @@ def get_ssl_cert(domain, host=None, port=443, timeout=3):
 
     ssl_context = ssl.create_default_context()
 
-    with ssl_context.wrap_socket(sock, server_hostname=domain) as wrap_socket:
-        return wrap_socket.getpeercert()
+    # fix: Python2 AttributeError: __exit__
+    wrap_socket = ssl_context.wrap_socket(sock, server_hostname=domain)
+    cert = wrap_socket.getpeercert()
+    wrap_socket.close()
+
+    return cert
 
 
 def get_ssl_cert_info(domain, host=None, port=443, timeout=3):
@@ -91,7 +95,7 @@ def resolve_cert(cert):
 
 
 if __name__ == '__main__':
-    # print(get_ssl_cert_info('www.taobao.com', '111.62.93.139'))
-    print(get_ssl_cert_info('38.60.47.102', '38.60.47.102'))
+    print(get_ssl_cert_info('www.taobao.com', '111.62.93.139'))
+    # print(get_ssl_cert_info('38.60.47.102', '38.60.47.102'))
     # print('www.baidu.com'.encode('idna')) # b'www.baidu.com'
     # print('www.baidu.com'.encode('punycode')) # b'www.baidu.com-'

domain_admin/utils/open_api/crtsh_api.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+"""
+@File    : crtsh_api.py
+@Date    : 2023-07-10
+
+参考:
+https://crt.sh/
+https://github.com/PaulSec/crt.sh
+
+需求：https://github.com/mouday/domain-admin/issues/41
+"""
+
+from __future__ import print_function, unicode_literals, absolute_import, division
+import requests
+
+USER_AGENT = 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:40.0) Gecko/20100101 Firefox/40.1'
+
+
+def search(domain):
+    """
+    搜索子域证书列表
+    :param domain: str 顶级域名
+    :return:
+    """
+    url = "https://crt.sh/"
+
+    params = {
+        'q': domain,
+        'output': 'json'
+    }
+
+    headers = {
+        'User-Agent': USER_AGENT
+    }
+
+    req = requests.get(url=url, params=params, headers=headers)
+
+    return req.json()
+
+
+if __name__ == '__main__':
+    lst = search('bilibili.com')
+    print([row['common_name'] for row in lst])

domain_admin/utils/time_util.py

@@ -4,12 +4,13 @@
 @Date    : 2023-06-03
 """
 from __future__ import print_function, unicode_literals, absolute_import, division
-from dateutil import parser
+
 from datetime import datetime
 
-# 时间格式化
-from peewee import DateTimeField
+from dateutil import parser
+from dateutil.tz import tzlocal
 
+# 时间格式化
 DATETIME_FORMAT = '%Y-%m-%d %H:%M:%S'
 
 
@@ -20,8 +21,9 @@ def parse_time(time_str):
     :return: datetime
     """
 
+    # fix: Python2 TypeError: Required argument 'tz' (pos 1) not found
     return datetime.strptime(
-        parser.parse(time_str).astimezone().strftime(DATETIME_FORMAT),
+        parser.parse(time_str).astimezone(tzlocal()).strftime(DATETIME_FORMAT),
         DATETIME_FORMAT
     )
 

domain_admin/utils/whois_util/util.py

@@ -7,6 +7,7 @@
 
 import socket
 from os import path
+import io
 
 
 def parse_whois_raw(whois_raw):
@@ -21,9 +22,10 @@ def parse_whois_raw(whois_raw):
         if 'Record expires on' in row or 'Record created on' in row:
             row_split = row.split("on", maxsplit=1)
         elif ":" in row:
-            row_split = row.split(":", maxsplit=1)
+            # fix: Python2 split() takes no keyword arguments
+            row_split = row.split(":", 1)
         else:
-            row_split = row.split(" ", maxsplit=1)
+            row_split = row.split(" ", 1)
 
         if len(row_split) == 2:
             key, value = row_split
@@ -71,8 +73,8 @@ def load_whois_servers():
     }
     """
     dct = {}
-
-    with open(path.join(path.dirname(__file__), 'whois-servers.txt'), 'r') as f:
+    # fix：Python2 encoding error
+    with io.open(path.join(path.dirname(__file__), 'whois-servers.txt'), 'r', encoding='utf-8') as f:
         for line in f:
             if line.startswith(";"):
                 pass

domain_admin/utils/whois_util/whois_util.py

@@ -221,5 +221,5 @@ def get_domain_info(domain):
 
 
 if __name__ == '__main__':
-    ret = get_domain_info('dot.ml')
+    ret = get_domain_info('baidu.com')
     print(ret)