import boto3 import csv import os import requests import json import time import re from collections import defaultdict from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.software_catalog_api import SoftwareCatalogApi OUTPUT_CSV = "lambda_service_catalog.csv" GITHUB_TOKEN = os.getenv("GITHUB_TOKEN", "dummy") GITHUB_ORG = "theorchard" JENKINS_ORG_URL = "https://pipeline.theorchard.io/job/theorchard/api/json" JENKINS_USER = os.getenv("JENKINS_USER", "dummy") JENKINS_API_TOKEN = os.getenv("JENKINS_API_TOKEN", "dummy") ENVIRONMENTS = ["dev", "qa", "stage", "uat", "prod", "shared", "backup", "networking", "unknown"] iam_client = boto3.client('iam') sts_client = boto3.client('sts') ssm_client = boto3.client('ssm') dd_config = Configuration() REPO_OVERRIDES_PATH = os.path.join(os.path.dirname(__file__), 'lambda_repo_overrides.json') if os.path.exists(REPO_OVERRIDES_PATH): with open(REPO_OVERRIDES_PATH, 'r') as f: REPO_OVERRIDES = json.load(f) else: REPO_OVERRIDES = {} def get_client(username, account, service): if account['name'] != 'prod': assume_role_response = sts_client.assume_role( RoleArn=f"arn:aws:iam::{account['account_id']}:role/{account['role']}", RoleSessionName=username ) credentials = assume_role_response["Credentials"] print(f"Assumed role for account {account['account_id']} ({account['name']}) successfully.") return boto3.client( service, aws_access_key_id=credentials["AccessKeyId"], aws_secret_access_key=credentials["SecretAccessKey"], aws_session_token=credentials["SessionToken"], ) else: return boto3.client(service) def get_aws_accounts(): aws_accounts = [] for page in ssm_client.get_paginator('describe_parameters').paginate( Shared=True, ParameterFilters=[ { 'Key': 'Name', 'Option': 'BeginsWith', 'Values': [ '/shared/aws-account-ids/', ] }, ] ): get_parameters_response = ssm_client.get_parameters( Names=[param['ARN'] for param in page['Parameters']], ) aws_accounts.extend([json.loads(parameter['Value']) for parameter in get_parameters_response['Parameters']]) return aws_accounts def get_repo_from_commit(commit_sha): url = f"https://api.github.com/search/commits?q={commit_sha}+org:{GITHUB_ORG}" headers = { "Accept": "application/vnd.github.cloak-preview", "Authorization": f"token {GITHUB_TOKEN}" } print(f"GitHub commit search URL: {url}") print(f"GitHub commit search headers: {headers}") max_retries = 10 retry_counter = 0 do_retry = True while do_retry and retry_counter < max_retries: resp = requests.get(url, headers=headers) print(f"GitHub commit search status: {resp.status_code}") print(f"GitHub commit search headers: {resp.headers}") print(f"GitHub commit search response: {resp.text}") # Throttle to avoid hitting the 30 req/min rate limit time.sleep(2) if resp.status_code == 200: items = resp.json().get("items", []) if items: return items[0]["repository"]["full_name"] if "You have exceeded a secondary rate limit." in resp.json().get("message", ""): retry_counter += 1 time.sleep(60) else: do_retry = False time.sleep(2) return "" def is_tag_commit_sha(s): """ Check if the input string is a valid Git commit SHA-1 hash. A valid SHA is a 40-character hexadecimal string. """ return bool(re.fullmatch(r'[0-9a-f]{40}', s)) def get_repo_from_lambda_function(lambda_client, function_name): try: response = lambda_client.get_function(FunctionName=function_name) code = response.get("Code", {}) image_uri = code.get("ImageUri", "") print(f"Function: {function_name}, ImageUri: {image_uri}") # Try to extract commit/tag from ECR image and search GitHub if image_uri and ":" in image_uri: _, tag = image_uri.rsplit(":", 1) if tag and is_tag_commit_sha(tag): print(f"Searching GitHub for commit/tag: {tag}") repo = get_repo_from_commit(tag) if repo: print(f"Found repo from commit/tag {tag}: {repo}") return repo else: print(f"No repo found for commit/tag {tag}") # Fallback: guess repo name from function name (remove env prefix/suffix) repo_candidate = function_name for env in ENVIRONMENTS: prefix = f"{env}-" suffix = f"-{env}" if repo_candidate.startswith(prefix): repo_candidate = repo_candidate[len(prefix):] if repo_candidate.endswith(suffix): repo_candidate = repo_candidate[:-len(suffix)] url = f"https://api.github.com/repos/{GITHUB_ORG}/{repo_candidate}" headers = {"Authorization": f"token {GITHUB_TOKEN}"} print(f"Calling GitHub API: {url}") print(f"Headers: {headers}") resp = requests.get(url, headers=headers) print(f"GitHub API response status: {resp.status_code}") print(f"GitHub API response content: {resp.text}") if resp.status_code == 200: print(f"Repo found by fallback: {GITHUB_ORG}/{repo_candidate}") return f"{GITHUB_ORG}/{repo_candidate}" else: print(f"Repo not found by fallback: {GITHUB_ORG}/{repo_candidate}") return "" except Exception as e: print(f"Error in get_repo_from_lambda_function: {e}") return "" def has_jenkinsfile(repo_full_name, path=""): if not repo_full_name: return False url = f"https://api.github.com/repos/{repo_full_name}/contents/{path}" if path else f"https://api.github.com/repos/{repo_full_name}/contents/" headers = {"Authorization": f"token {GITHUB_TOKEN}"} resp = requests.get(url, headers=headers) if resp.status_code != 200: return False files = resp.json() if isinstance(files, dict) and files.get("type") == "file": return files.get("name") == "Jenkinsfile" return any(f.get("name") == "Jenkinsfile" for f in files if isinstance(f, dict)) def is_in_jenkins_org_folder(repo_name): try: auth = (JENKINS_USER, JENKINS_API_TOKEN) if JENKINS_USER and JENKINS_API_TOKEN else None resp = requests.get(JENKINS_ORG_URL, auth=auth) if resp.status_code != 200: return False jobs = resp.json().get("jobs", []) return any(job.get("name") == repo_name for job in jobs) except Exception: return False def get_logical_function_name(function_name): for env in ENVIRONMENTS: prefix = f"{env}-" suffix = f"-{env}" if function_name.startswith(env) or function_name.endswith(env): return function_name.removeprefix(prefix).removesuffix(suffix) return function_name def get_lambda_tags(lambda_client, function_arn): try: tags = lambda_client.list_tags(Resource=function_arn).get("Tags", {}) return tags.get("service_name"), tags.get("application_family") except Exception: return None, None def set_app_family(logical_name): if logical_name.startswith("apollo-"): return "apollo" if logical_name.startswith("delphi-"): return "delphi" if logical_name.startswith("artistapp-") or logical_name.startswith("rti-"): return "rti" return "" def security_lambdas_repo(name): if name.endswith("notify_security") or \ name.endswith("lookup_cloudtrail_events") or \ name.endswith("delete_access_key_pair"): return "theorchard/terraform-aws-risk-credentials-exposed" return "" def skip_lambda(env_accounts): """Returns true if lambda envs are either dev or unknown and account id is dev account""" excluded_env_set = set(["dev", "unknown",]) envs = [env for env in env_accounts.keys() if len(env_accounts[env]) > 0] if set(envs).issubset(excluded_env_set): account_ids = set([id for ids in env_accounts.values() for id in ids]) if account_ids == set(["103233932089"]): return True return False def is_in_software_catalog(service_name): with ApiClient(dd_config) as api_client: api_instance = SoftwareCatalogApi(api_client) response = api_instance.list_catalog_entity(filter_name=service_name) for entity in response["data"]: attributes = entity.get("attributes", {}) if attributes.get("kind") == "service" and 'metadata_origin:datadog-tools' in attributes.get("tags", []): return True return False def main(): current_user = iam_client.get_user() username = current_user['User']['UserName'] # {(app_family, service_name, logical_name): {env: [account_ids]}} functions_info = defaultdict(lambda: {env: [] for env in ENVIRONMENTS}) repo_map = {} for account in get_aws_accounts(): print(f"Processing AWS account: {account}") try: sts = get_client(username, account, "sts") lambda_client = get_client(username, account, "lambda") aws_account_id = sts.get_caller_identity()["Account"] paginator = lambda_client.get_paginator("list_functions") for page in paginator.paginate(): for fn in page["Functions"]: function_name = fn["FunctionName"] function_arn = fn["FunctionArn"] logical_name = get_logical_function_name(function_name) environment = "unknown" for env in ENVIRONMENTS: if function_name.startswith(f"{env}-") or function_name.endswith(f"-{env}"): environment = env break service_name, app_family = get_lambda_tags(lambda_client, function_arn) if not service_name: service_name = logical_name if not app_family: app_family = set_app_family(logical_name) key = (app_family, service_name, logical_name) if aws_account_id not in functions_info[key][environment]: functions_info[key][environment].append(aws_account_id) if key not in repo_map or not repo_map[key]: # Check manual override if repo not found if REPO_OVERRIDES.get(service_name): repo_full_name = REPO_OVERRIDES.get(service_name, "") print(f"Using manual repo override for {service_name}: {repo_full_name}") else: repo_full_name = security_lambdas_repo(function_name) if not repo_full_name: repo_full_name = get_repo_from_lambda_function(lambda_client, function_name) repo_map[key] = repo_full_name except Exception as e: print(f"Failed for account {account}: {e}") continue with open(OUTPUT_CSV, "w", newline="") as csvfile: writer = csv.writer(csvfile) writer.writerow([ "application_family", "github_repo", "service_name", "function_name", "dev", "qa", "stage", "uat", "prod", "shared", "backup", "networking", "unknown", "has_jenkinsfile", "in_jenkins_org_folder", "in_software_catalog", ]) for (app_family, service_name, logical_name), env_accounts in functions_info.items(): if skip_lambda(env_accounts): print(f"skipping {(app_family, service_name, logical_name)}") else: print((app_family, service_name, logical_name)) repo_full_name = repo_map.get((app_family, service_name, logical_name), "") repo_name = repo_full_name.split("/")[-1] if repo_full_name else "" has_jenkins = has_jenkinsfile(repo_full_name) in_jenkins_org = is_in_jenkins_org_folder(repo_name) in_software_catalog = is_in_software_catalog(service_name) row = [ app_family or "", repo_full_name or "", service_name or "", logical_name or "", ] for env in ENVIRONMENTS: account_ids = env_accounts.get(env, []) row.append(";".join(account_ids) if account_ids else "") row.extend([str(has_jenkins), str(in_jenkins_org), str(in_software_catalog)]) writer.writerow(row) print(f"Done! CSV written to {OUTPUT_CSV}") if __name__ == "__main__": main()