晨曦's Blog

import requests
import hashlib
import time

key = "xxx" # 密钥，由服务端颁发

client = "local" # 客户端
server = "kafka" # 服务端
env = "test" # 环境
timestamp = str(time.time())[:10] # 时间
sign_str = client+server+env+timestamp+key
sign = hashlib.md5(sign_str.encode(encoding='utf-8')).hexdigest() # md5

headers = dict(client=client, server=server, env=env, timestamp=timestamp, sign=sign)

r1 = requests.get("https://www.xxx.com/hello", headers=headers)

from sanic import Sanic
from sanic.response import text, json
import hashlib

app = Sanic("My Hello, world app")

@app.get("/hello")
async def hello_world(request):

      key = "xxx"  # 密钥

      client = request.headers.get("client")
      server = request.headers.get("server")
      env = request.headers.get("env")
      timestamp = request.headers.get("timestamp")
      sign = request.headers.get("sign")

      _sign_str = client+server+env+timestamp+key
      _sign = hashlib.md5(_sign_str.encode(encoding='utf-8')).hexdigest()

      if sign==_sign:
            return json(list(request.headers.items()))
      else:
            return text("验证失败")
app.run(host='0.0.0.0',debug=True,auto_reload=True)

./server.py

from sanic import Sanic
from databases import Database

app = Sanic('jakehu')

# 数据库
database = Database('mysql://localhost/example')
# 服务开启之前
@app.listener("before_server_start")
async def startup_db(app, loop):
    await database.connect()
# 服务结束之后
@app.listener("after_server_stop")
async def shutdown_db(app, loop):
    await database.disconnect()
app.ctx.db = database

@app.get("/")
async def foo_handler(request):
    return text("Hello World!")

app.run(host='0.0.0.0', debug=True, auto_reload=True)

@app.get("/")
async def foo_handler(request):
    database= request.app.ctx.db
    query = "SELECT * FROM table"
    rows = await database.fetch_all(query=query)
    result = [dict(row._mapping) for row in rows]
    return json(result)

@app.route("/upload", methods=['POST'])
async def upload(request):
    allow_type = ['.jpg', '.png', '.gif'] # 允许上传的类型
    file = request.files.get('file')
    type = os.path.splitext(file.name)

    if type[1] not in allow_type:
        return json({"code": -1, "msg": "只允许上传.jpg.png.gif类型文件"})

    name = str(uuid4())+type[1]
    path = "/user/data/web/upload" # 这里注意path是绝对路径

    async with aiofiles.open(path+"/"+name, 'wb') as f:
        await f.write(file.body)
    f.close()
    return json({"code": 0, "msg": "上传成功", "data": {
        "name": name,
        "url": "/upload/"+name
    }})

path = "/user/data/web/upload" # 这里注意path是绝对路径
app.static("/upload", path)

from tortoise import Model, fields
from uuid import uuid4

class ag_ip_group(Model):
    id = fields.IntField(pk=True)
    uuid = fields.UUIDField(default=uuid4)
    name = fields.CharField(64)
    ip = fields.JSONField()
    remark = fields.CharField(64,default='_')
    status = fields.IntField(default=1)
    created_time = fields.DatetimeField(null=True, auto_now_add=True)
    updated_time = fields.DatetimeField(null=True, auto_now=True)

    def __str__(self):
        return str(self.id)

from tortoise.functions import Function
from pypika import CustomFunction

class JsonContains(Function):
    database_func = CustomFunction("JSON_CONTAINS", ["field", "value"])

filter = {}
annotate = {}
annotate['json_len'] = JsonContains('ip', '"'+row['ip']+'"')
filter['json_len__gt'] = 0

await self.model.all().annotate(**annotate).filter(**filter).limit(limit).offset(offset).values()

class JsonExtract(Function):
    database_func = CustomFunction("JSON_EXTRACT", ["field", 'value'])

filter = {}
annotate = {}
annotate['json_str'] = JsonExtract('ip','$')
filter['json_str__icontains'] = row['ip']

await self.model.all().annotate(**annotate).filter(**filter).limit(limit).offset(offset).values()

import json
field = fields.JSONField(encoder=lambda x: json.dumps(x,ensure_ascii=False))

from OpenSSL import crypto
from dateutil import parser

class Ssl():
    def parse(self, certificate):
        res = {
            "domain": []
        }
        cert = crypto.load_certificate(crypto.FILETYPE_PEM, certificate)
        # 有效期
        datetime_struct_start = parser.parse(cert.get_notBefore().decode("UTF-8"))
        datetime_struct_end = parser.parse(cert.get_notAfter().decode("UTF-8"))
        res['start_time'] = datetime_struct_start.strftime('%Y-%m-%d %H:%M:%S')
        res['end_time'] = datetime_struct_end.strftime('%Y-%m-%d %H:%M:%S')
        # 扩展
        count = cert.get_extension_count()
        for i in range(count):
            crt = cert.get_extension(i)
            name = str(crt.get_short_name(), encoding="utf-8")
            if name == "subjectAltName":
                _list = str(crt).split(',')
                _it = iter(_list)    # 创建迭代器对象
                for x in _it:
                    res["domain"].append(x.strip().replace("DNS:", ""))
                break
        return res

if __name__ == '__main__':
    _certificate="""证书私钥"""
    ssl = Ssl()
    print(ssl.parse(_certificate))
    # {'domain': ['*.jakehu.me', 'jakehu.me'], 'start_time': '2021-07-12 18:01:16', 'end_time': '2021-10-10 18:01:15'}

subject = cert.get_subject()
issuer = cert.get_issuer()

print("证书版本:", cert.get_version() + 1)
print("证书序列号:", hex(cert.get_serial_number()))
print("证书中使用的签名算法:", cert.get_signature_algorithm().decode("UTF-8"))
print("颁发者:", issuer.commonName)
datetime_struct = parser.parse(cert.get_notBefore().decode("UTF-8"))
print("有效期从:", datetime_struct.strftime('%Y-%m-%d %H:%M:%S'))
datetime_struct= parser.parse(cert.get_notAfter().decode("UTF-8"))
print("到:", datetime_struct.strftime('%Y-%m-%d %H:%M:%S'))
print("证书是否已经过期:", cert.has_expired())
print("公钥长度", cert.get_pubkey().bits())
print("公钥:\n", crypto.dump_publickey(crypto.FILETYPE_PEM, cert.get_pubkey()).decode("utf-8"))
print("主体信息:")
print("CN:通用名称  OU:机构单元名称")
print("O:机构名  L:地理位置")
print("S:州/省名  C:国名")
for item in issuer.get_components():
    print(item[0].decode("utf-8"), "  ——  ", item[1].decode("utf-8"))