#### # detect the leaked gcp vm instance on dev environments # it will be added into gcloud function later ### import base64 import csv import hmac import json import time import os import threading from datetime import datetime from hashlib import sha256 import google.auth import googleapiclient import requests as httpRequests import yaml import googleapiclient from googleapiclient import discovery from google.auth.transport import requests from kubernetes import client, config from oauth2client.client import GoogleCredentials gkeClusters = { "staging-seed-asia-northeast1-83d8caa": "asia-northeast1", "staging-seed-asia-southeast1-b7bc5b9": "asia-southeast1", "staging-seed-us-central1-6159679": "us-central1", "dev-seed-us-west1-3137aff": "us-west1", "stability-seed-us-west1-67ac24a": "us-west1", "staging-seed-us-west1-73cb72c": "us-west1", "dev-seed-us-west2-9f7b847": "us-west2" } gkeAddresses = { "staging-seed-asia-northeast1-83d8caa": "35.190.235.134", "staging-seed-asia-southeast1-b7bc5b9": "35.240.161.172", "staging-seed-us-central1-6159679": "35.223.148.203", "dev-seed-us-west1-3137aff": "34.82.26.237", "stability-seed-us-west1-67ac24a": "35.227.149.173", "staging-seed-us-west1-73cb72c": "35.227.149.173", "dev-seed-us-west2-9f7b847": "35.236.101.158" } global leakedNumer leakedNumer = 0 # feishu_webhook = 'https://open.feishu.cn/open-apis/bot/v2/hook/402619ca-3aad-4d9f-86a4-8e1bea90bb76' feishu_webhook = os.getenv("feishu_webhook") feishu_secret = os.getenv("feishu_secret") outfile = 'disks-{}.csv'.format(datetime.now().strftime('%Y-%m-%d-%H:%M:%S')) bucketName = 'daily-resource-leak-detetor' csvFile = open('/tmp/' + outfile, 'w', newline='') fieldnames = ['id', 'name', 'zone', 'machineType', 'creationTimestamp', 'selfLink'] writer = csv.DictWriter(csvFile, fieldnames=fieldnames) writer.writeheader() instances = [] allRunInstances = [] def get_all_instances(): computeClient = googleapiclient.discovery.build('compute', 'v1') credentials = GoogleCredentials.get_application_default() instancesRequest = computeClient.instances().aggregatedList(project="pingcap-gardener") while instancesRequest is not None: instanceResponse = instancesRequest.execute() for name, instances_scoped_list in instanceResponse['items'].items(): if 'instances' in instances_scoped_list and len(instances_scoped_list['instances']) > 0: for instance in instances_scoped_list['instances']: instances.append( [instance['id'], instance['name'], instance['zone'], instance['selfLink'], instance['machineType'], instance['creationTimestamp']]) instancesRequest = computeClient.instances().aggregatedList_next(previous_request=instancesRequest, previous_response=instanceResponse) def get_gke_nodes(k8sIp): credentials, project_id = google.auth.default() session = requests.AuthorizedSession(credentials) response = session.get('https://{}/api/v1/nodes'.format(k8sIp), verify=False) response.raise_for_status() return response.json() def read_namespaces(k8sIp): credentials, project_id = google.auth.default() session = requests.AuthorizedSession(credentials) response = session.get('https://{}/api/v1/namespaces'.format(k8sIp), verify=False) response.raise_for_status() return response.json() def process_gke_clusters(): for gke in gkeClusters: nodes = get_gke_nodes(gkeAddresses[gke]) for node in nodes['items']: print("process gke {} node {}".format(gke, node['metadata']['name'])) allRunInstances.append(node['metadata']['name']) def read_kubeconfig_secret(k8sIp, ns, secret): credentials, project_id = google.auth.default() session = requests.AuthorizedSession(credentials) response = session.get('https://{}/api/v1/namespaces/{}/secrets/{}'.format(k8sIp, ns, secret), verify=False) response.raise_for_status() return response.json() def process_shoot_cluster(gke, shootNs): try: secret = read_kubeconfig_secret(gkeAddresses[gke], shootNs['metadata']['name'], 'kubecfg') shootKubeconfig = secret['data']["kubeconfig"] shootKubeconfig = base64.b64decode(shootKubeconfig) shootKubeconfigFile = open("/tmp/" + shootNs['metadata']['name'], "w") yaml.dump(yaml.load(shootKubeconfig), shootKubeconfigFile) shootKubeconfigFile.close() config.kube_config.load_kube_config(config_file="/tmp/" + shootNs['metadata']['name']) apiClient = client.CoreV1Api() nodes = apiClient.list_node() namespaces = apiClient.list_namespace() pods = [] for ns in namespaces.items: namespaced_pods = apiClient.list_namespaced_pod(ns.metadata.name) for pod in namespaced_pods.items: pods.append(pod) for node in nodes.items: print(shootNs) print("process shoot {} node {}".format(shootNs['metadata']['name'], node.metadata.name)) if "master" in node.metadata.name: allRunInstances.append(node.metadata.name) nodeFoundIt = False for pod in pods: if 'extra' in pod.metadata.name or \ 'db-pd' in pod.metadata.name or \ 'db-tidb' in pod.metadata.name or \ 'db-tikv' in pod.metadata.name or \ 'db-discovery' in pod.metadata.name or \ 'db-monitor' in pod.metadata.name: try: if pod.spec.node_name == node.metadata.name: nodeFoundIt = True break except Exception as e: print("bypass pod {}, because of {}".format(pod.metadata.name, str(e))) if nodeFoundIt: allRunInstances.append(node.metadata.name) except Exception as e: print("bypass shoot {}, because of {}".format(shootNs['metadata']['name'], str(e))) shoot_threads = [] def process_all_shoot_clusters(): for gke in gkeClusters: namespaces = read_namespaces(gkeAddresses[gke]) print(namespaces) for shootNs in namespaces['items']: if 'shoot--' not in shootNs['metadata']['name']: continue print("process shoot {}".format(shootNs['metadata']['name'])) t = threading.Thread(target=process_shoot_cluster, args=(gke, shootNs)) shoot_threads.append(t) t.start() for t in shoot_threads: t.join() def process(): for instance in allRunInstances: print(instance) for instance in instances: foundIt = False for runInstance in allRunInstances: if runInstance == instance[1]: foundIt = True seps = str.split(instance[4], '/') machineType = seps[len(seps) - 1] seps = str.split(instance[2], '/') region = seps[len(seps) - 1] if not foundIt: global leakedNumer leakedNumer = leakedNumer + 1 writer.writerow({ 'id': instance[0], 'name': instance[1], 'machineType': machineType, 'creationTimestamp': instance[5], 'zone': region, 'selfLink': instance[3]}) csvFile.close() def get_sign_timestamp(): timestamp = str(int(time.time())) # secret = os.getenv("feishu_secret") key = f'{timestamp}\n{feishu_secret}' key_enc = key.encode('utf-8') msg = "" msg_enc = msg.encode('utf-8') hmac_code = hmac.new(key_enc, msg_enc, digestmod=sha256).digest() sign = base64.b64encode(hmac_code).decode('utf-8') return {'timestamp': timestamp, 'sign': sign} def post_bot_webhook(): sign_timestamp = get_sign_timestamp() payload_message = { "timestamp": sign_timestamp['timestamp'], "sign": sign_timestamp['sign'], "msg_type": "interactive", "card": { "i18n_elements": { "en_us": [ { "tag": "div", "text": { "tag": "lark_md", "content": "**可能泄露的GCP VM有{}台**".format(leakedNumer) }, }, { "tag": "div", "text": { "tag": "lark_md", "content": "**结果存放在gcp对象存储：{}/{} **".format(bucketName, outfile) }, }, ] } } } headers = { 'Content-Type': 'application/json' } response = httpRequests.request( "POST", feishu_webhook, headers=headers, data=json.dumps(payload_message)) print(response.text) def upload_result_to_gcp_storage_and_send_alerts(): gcpBlobClient = googleapiclient.discovery.build('storage', 'v1') body = { 'name': outfile, } with open('/tmp/' + outfile, 'rb') as f: req = gcpBlobClient.objects().insert( bucket='daily-resource-leak-detetor', body=body, media_body=googleapiclient.http.MediaIoBaseUpload( f, 'application/octet-stream')) resp = req.execute() post_bot_webhook() get_all_instances() process_gke_clusters() process_all_shoot_clusters() process() upload_result_to_gcp_storage_and_send_alerts()