import base64 import datetime import hashlib import hmac import json import logging import os import socket import ssl import time from collections import defaultdict from distutils.command.config import config import requests from prometheus_client import start_http_server, Counter import schedule import yaml from Crypto.Cipher import AES from Crypto.Util.Padding import pad from requests.auth import HTTPDigestAuth CONFIG_PATH = os.getenv("CONFIG_PATH") config = yaml.safe_load(open(CONFIG_PATH)) FEISHU_WEBHOOK_URL = config["feishu"]["webhook_url"] FEISHU_SIGN_SECRET = config["feishu"]["sign_secret"] # create logger logger = logging.getLogger("simple_example") logger.setLevel(logging.DEBUG) # create console handler and set level to debug ch = logging.StreamHandler() ch.setLevel(logging.DEBUG) # create formatter formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") # add formatter to ch ch.setFormatter(formatter) # add ch to logger logger.addHandler(ch) prom_probe_result = Counter("kong_sentinel_probe_result", "", ["domain", "code"]) def gen_sign(timestamp, secret): # Concatenate timestamp and secret string_to_sign = "{}\n{}".format(timestamp, secret) hmac_code = hmac.new( string_to_sign.encode("utf-8"), digestmod=hashlib.sha256 ).digest() # Encode the result with Base64 sign = base64.b64encode(hmac_code).decode("utf-8") return sign def send_alert(content): now = str(int(time.time())) sign = gen_sign(now, FEISHU_SIGN_SECRET) data = { "timestamp": now, "sign": sign, "msg_type": "text", "content": {"text": content}, } resp = requests.post(FEISHU_WEBHOOK_URL, json=data, timeout=3) logger.info("send alert result: {}".format(resp.json())) class ErrorStatus: def __init__(self, domain, msg_type, interval=60, threshold=3) -> None: self.domain = domain self.status = defaultdict(int) self.last_send = None self.first_send = None self.interval = interval self.threshold = threshold self.msg_type = msg_type def incr(self, code): self.status[code] += 1 def clear(self): if len(self.status) > 0: if self.last_send is not None: self.send(recover=True) self.status = defaultdict(int) self.last_send = None self.first_send = None def try_send(self): if ( self.last_send is not None and datetime.datetime.now() - self.last_send < datetime.timedelta(seconds=self.interval) ): return self.send() def send(self, recover=False): codes = [] for code, count in self.status.items(): if count >= self.threshold: codes.append("{}: {} 次".format(code, count)) if len(codes) == 0: return lasts = ( (datetime.datetime.now() - self.first_send).seconds if self.first_send is not None else 0 ) if recover: content = "{} {}恢复，持续时间 {} 秒".format(self.domain, self.msg_type, lasts) else: content = "{} {}失败持续 {} 秒\n{}".format( self.domain, self.msg_type, lasts, "\n".join(codes) ) send_alert(content) self.last_send = datetime.datetime.now() if self.first_send is None: self.first_send = self.last_send class HealthySentinel: def __init__(self, domains): self.domains = domains result = {} for domain in domains: result[domain] = ErrorStatus(domain, "健康检查") self.result = result def run(self): for domain in self.domains: error_status = self.result[domain] success, code = self.check(domain) prom_probe_result.labels(domain, code).inc() if success: error_status.clear() else: error_status.incr(code) error_status.try_send() def check(self, domain): url = "https://{}/status/200".format(domain) try: resp = requests.get(url, timeout=3, headers={"kong-debug": "1"}) except requests.RequestException as e: code = "HEALTHY:HTTP_CALL_EXCEPTION" logger.error("{}, code: {}, err: {}".format(domain, code, e)) return False, code if resp.status_code != 200: code = "HEALTHY:UNEXPECTED_STATUS" logger.error( "{}, code: {}, err: expected 200 but got {}".format( domain, code, resp.status_code ) ) return False, code if not resp.headers.get("Kong-Service-Name", "").startswith("kong.go-httpbin"): code = "HEALTHY:UNEXPECTED_HEADER" logger.error( "{}, code: {}, err: expected go-httpbin but got {}".format( domain, code, resp.headers.get("Kong-Service-Name") ) ) return False, code return True, "HEALTHY:SUCCESS" class AuthDomain: def __init__(self, domain, username, password) -> None: self.domain = domain self.username = username self.password = password class AuthenticationSentinel: def __init__(self, auth_domains): self.auth_domains = auth_domains self.result = {} for auth_domain in auth_domains: self.result[auth_domain.domain] = ErrorStatus(auth_domain.domain, "认证检查") def run(self): for auth_domain in self.auth_domains: error_status = self.result[auth_domain.domain] success, code = self.check( auth_domain.domain, auth_domain.username, auth_domain.password ) prom_probe_result.labels(auth_domain.domain, code).inc() if success: error_status.clear() else: error_status.incr(code) error_status.try_send() @staticmethod def check(domain, username, password, expected=200): url = "https://{}/status/200".format(domain) auth = HTTPDigestAuth(username, password) try: resp = requests.get(url, timeout=3, auth=auth) except requests.RequestException as e: code = "AUTH:HTTP_CALL_EXCEPTION" logger.error("{}, code: {}, err: {}".format(domain, code, e)) return False, code if resp.status_code != expected: code = "AUTH:UNEXPECTED_STATUS" logger.error( "{}, code: {}, err: expected {} but got {}".format( domain, code, expected, resp.status_code ) ) return False, code return True, "AUTH:SUCCESS" class CertificateSentinel: def __init__(self, domains): self.domains = domains self.result = {} for domain in domains: self.result[domain] = ErrorStatus( domain, "证书检查", interval=60 * 60, threshold=3 ) def run(self): for domain in self.domains: error_status = self.result[domain] success, code = self.check(domain) prom_probe_result.labels(domain, code).inc() if success: error_status.clear() else: error_status.incr(code) error_status.try_send() def check(self, domain): try: context = ssl.create_default_context() conn = context.wrap_socket( socket.socket(socket.AF_INET), server_hostname=domain ) conn.settimeout(3) conn.connect((domain, 443)) cert = conn.getpeercert() except BaseException as e: code = "CERT:CALL_EXCEPTION" logger.error("{}, code:{}, err: {}".format(domain, code, e)) return False, code finally: conn.close() expire_at = ssl.cert_time_to_seconds(cert["notAfter"]) if expire_at - time.time() < 30 * 24 * 60 * 60: code = "CERT:EXPIRED" return False, code return True, "CERT:SUCCESS" class CrossRegionConfig: def __init__(self, write_domain, email, org_id, salt, read_domains): self.write_domain = write_domain self.email = email self.org_id = org_id self.salt = salt self.read_domains = read_domains class Driver: def __init__(self, config): self.config = config self._flush_token() self.public_key = "" self.private_key = "" self.key_id = 0 # generate a token for testing # refer to: https://github.com/tidbcloud/core-svc/blob/fb96b3cc99fc447cfed9603a2d115f2a7407e75a/portal/internal/central/model/testtoken/token.go#L44 def _generate_test_token(self): expire_at = int(time.time()) + 24 * 3600 data = json.dumps( {"Email": self.config.email, "ExpireTime": expire_at}, separators=(",", ":"), ) # data = '{"Email":"jiazhi.wen@pingcap.com","ExpireTime":1679712588}' # CLOUDoGfX5GO-3_Fg1HI80AP055VASPhF0GKUH2_ungkXmr3ltcswkfyrbXg5t_dHho0Gb-bVbLmIncg2G1BhWRY8vA== bytes_salt = self.config.salt.encode("utf-8") cipher = AES.new(bytes_salt, AES.MODE_CBC, iv=bytes_salt) ct_bytes = cipher.encrypt(pad(data.encode("utf-8"), AES.block_size)) token = "Bearer CLOUD" + base64.urlsafe_b64encode(ct_bytes).decode("utf-8") return token, expire_at def _flush_token(self): token, expire_at = self._generate_test_token() self.token = token self.expire_at = expire_at def check(self): self._check_token() if not self.public_key: success, code = self._check_create_key() prom_probe_result.labels(self.config.write_domain, code).inc() if not success: return False, code if self.public_key: success, code = self._check_delete_key() prom_probe_result.labels(self.config.write_domain, code).inc() if not success: return False, code return True, "" def _check_token(self): if time.time() - 3600 > self.expire_at: self._flush_token() def _check_create_key(self): try: resp = requests.post( f"https://{self.config.write_domain}/api/v1/orgs/{self.config.org_id}/api_keys", json={ "description": "Generated by cross region sentinel, DO NOT EDIT", "role": "owner", "org_id": f"{self.config.org_id}", }, headers={ "Authorization": self.token, }, timeout=10, ) except requests.RequestException as e: code = "CROSS_REGION:HTTP_CALL_EXCEPTION" logger.error( "{}, code: {}, err: {}".format(self.config.write_domain, code, e) ) return False, code if resp.status_code != 200: code = "CROSS_REGION:UNEXPECTED_STATUS" logger.error( "{}, code: {}, err: expected 200 but got {}".format( self.config.write_domain, code, resp.status_code ) ) self._flush_token() return False, code data = resp.json() self.key_id = data["id"] self.public_key = data["public_key"] self.private_key = data["private_key"] logger.info("api key created: {}".format(self.public_key)) # Wait 10 seconds for data sync time.sleep(10) for read_domain in self.config.read_domains: success, code = AuthenticationSentinel.check( read_domain, self.public_key, self.private_key ) prom_probe_result.labels(read_domain, code).inc() if not success: cross_code = "CROSS_REGION:REGION_FAIL" logger.error( "{}, cross_code: {}, read_domain: {}, code: {}".format( self.config.write_domain, cross_code, read_domain, code ) ) return False, cross_code return True, "CROSS_REGION:SUCCESS" def _check_delete_key(self): try: resp = requests.delete( f"https://{self.config.write_domain}/api/v1/orgs/{self.config.org_id}/api_keys/{self.key_id}", headers={ "Authorization": self.token, }, timeout=10, ) except requests.RequestException as e: code = "CROSS_REGION:HTTP_CALL_EXCEPTION" logger.error( "{}, code: {}, err: {}".format(self.config.write_domain, code, e) ) return False, code if resp.status_code != 200: code = "CROSS_REGION:UNEXPECTED_STATUS" logger.error( "{}, code: {}, err: expected 200 but got {}".format( self.config.write_domain, code, resp.status_code ) ) self._flush_token() return False, code logger.info("api key deleted: {}".format(self.public_key)) self.key_id = 0 self.public_key = "" self.private_key = "" # Wait 30 seconds for data sync time.sleep(30) for read_domain in self.config.read_domains: success, code = AuthenticationSentinel.check( read_domain, self.public_key, self.private_key, expected=401 ) prom_probe_result.labels(read_domain, code).inc() if not success: cross_code = "CROSS_REGION:REGION_FAIL" logger.error( "{}, cross_code: {}, read_domain: {}, code: {}".format( self.config.write_domain, cross_code, read_domain, code ) ) return False, cross_code return True, "CROSS_REGION:SUCCESS" class State: def __init__(self, config: CrossRegionConfig): self.config = config self.driver = Driver(config) self.error_status = ErrorStatus( config.write_domain, "CrossRegion检查", interval=3 * 60, threshold=2 ) def run(self): success, code = self.driver.check() if success: self.error_status.clear() else: self.error_status.incr(code) self.error_status.try_send() class CrossRegionSentinel: def __init__(self, configs): self.states = {} for config in configs: self.states[config.write_domain] = State(config) def run(self): for state in self.states.values(): state.run() def main(): h = HealthySentinel(config["healthy"]) schedule.every(6).seconds.do(h.run) a = AuthenticationSentinel( [ AuthDomain(c["domain"], c["username"], c["password"]) for c in config["authentication"] ] ) schedule.every(6).seconds.do(a.run) c = CertificateSentinel(config["certificate"]) schedule.every(10 * 60).seconds.do(c.run) cr = CrossRegionSentinel( [ CrossRegionConfig( c["write_domain"], c["email"], c["org_id"], c["salt"], c["read_domains"] ) for c in config["cross_region"] ] ) # schedule.every(3 * 60).seconds.do(cr.run) start_http_server(9090) while True: schedule.run_pending() time.sleep(1) if __name__ == "__main__": main()