"""Raio-X de suporte estendido da AWS — coletor somente leitura (Observei). Lista versões e tamanhos de RDS/Aurora, ElastiCache, OpenSearch, EKS e instâncias EC2 com Amazon Linux 2, para calcular quanto a conta paga (ou vai pagar) de suporte estendido. O que este script faz: só chamadas de leitura à AWS (Describe/List/Get) e grava um arquivo JSON na pasta atual. O que ele NÃO faz: não lê dados dos bancos, logs ou segredos; não cria, altera nem apaga nada; não envia nada pela internet. Você decide se e para quem mandar o arquivo. Uso no AWS CloudShell (boto3 já vem instalado): curl -fsSLO https://observei.vercel.app/raio-x/coletar.py python3 coletar.py --cliente "Sua empresa" python3 coletar.py --cliente "Sua empresa" --anonimizar # troca os nomes dos recursos por apelidos python3 coletar.py --cliente "Sua empresa" --sem-custos # não consulta o Cost Explorer (US$ 0,01) Uso interno da Observei (role somente leitura ou perfil local): python3 coletar.py --role-arn arn:aws:iam::123456789012:role/observei-raio-x-readonly --external-id python3 coletar.py --perfil meu-perfil --regioes sa-east-1 us-east-1 """ import argparse import datetime as dt import json import os import sys from collections import defaultdict try: import boto3 from botocore.config import Config from botocore.exceptions import ClientError, ParamValidationError except ImportError: sys.exit("Este script precisa do boto3 (já vem instalado no AWS CloudShell). Fora dele: pip install boto3") VERSAO = "2026-10-01" CONTATO = "WhatsApp (11) 99494-6744 ou guigobrito@hotmail.com" CFG = Config(retries={"max_attempts": 8, "mode": "adaptive"}) avisos = [] def descrever_erro(e): """Só o código e a operação: a mensagem completa da AWS pode trazer ARNs e nomes internos.""" if isinstance(e, ClientError): return f"{e.response.get('Error', {}).get('Code', 'erro')} em {e.operation_name}" return e.__class__.__name__ def aviso(msg): avisos.append(msg) print(f" ! {msg}", file=sys.stderr) def sessao(a): base = boto3.Session(profile_name=a.perfil) if a.perfil else boto3.Session() if not a.role_arn: return base kw = {"RoleArn": a.role_arn, "RoleSessionName": "observei-raio-x"} if a.external_id: kw["ExternalId"] = a.external_id c = base.client("sts").assume_role(**kw)["Credentials"] return boto3.Session(aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"], aws_session_token=c["SessionToken"]) def paginar(cliente, metodo, chave, **kw): for pagina in cliente.get_paginator(metodo).paginate(**kw): yield from pagina.get(chave, []) def coletar_rds(s, regiao, saida): rds = s.client("rds", region_name=regiao, config=CFG) for i in paginar(rds, "describe_db_instances", "DBInstances"): saida.append({ "tipo": "rds", "regiao": regiao, "id": i["DBInstanceIdentifier"], "engine": i["Engine"], "versao": i["EngineVersion"], "classe": i["DBInstanceClass"], "multi_az": bool(i.get("MultiAZ")), "cluster": i.get("DBClusterIdentifier"), "replica_de": i.get("ReadReplicaSourceDBInstanceIdentifier"), "lifecycle": i.get("EngineLifecycleSupport"), "status": i.get("DBInstanceStatus"), }) def coletar_elasticache(s, regiao, saida): ec = s.client("elasticache", region_name=regiao, config=CFG) for c in paginar(ec, "describe_cache_clusters", "CacheClusters"): saida.append({ "tipo": "elasticache", "regiao": regiao, "id": c["CacheClusterId"], "grupo": c.get("ReplicationGroupId") or c["CacheClusterId"], "engine": c["Engine"], "versao": c["EngineVersion"], "classe": c["CacheNodeType"], "nos": c.get("NumCacheNodes", 1), }) def coletar_opensearch(s, regiao, saida): os_ = s.client("opensearch", region_name=regiao, config=CFG) nomes = [d["DomainName"] for d in os_.list_domain_names().get("DomainNames", [])] for lote in (nomes[i:i + 5] for i in range(0, len(nomes), 5)): for d in os_.describe_domains(DomainNames=lote)["DomainStatusList"]: cc = d.get("ClusterConfig", {}) instancias = [{"classe": cc.get("InstanceType"), "qtd": cc.get("InstanceCount", 0)}] if cc.get("DedicatedMasterEnabled"): instancias.append({"classe": cc.get("DedicatedMasterType"), "qtd": cc.get("DedicatedMasterCount", 0)}) if cc.get("WarmEnabled"): instancias.append({"classe": cc.get("WarmType"), "qtd": cc.get("WarmCount", 0)}) saida.append({"tipo": "opensearch", "regiao": regiao, "id": d["DomainName"], "versao": d.get("EngineVersion"), "instancias": instancias}) def versoes_eks(eks, regiao): """Status de suporte de cada versão do Kubernetes na região (API recente: pode faltar em boto3 antigo).""" status, kw = {}, {"includeAll": True} try: while True: try: resp = eks.describe_cluster_versions(**kw) except ParamValidationError: if "includeAll" not in kw: raise kw.pop("includeAll") # versões antigas do boto3 não conhecem o parâmetro continue for v in resp.get("clusterVersions", []): status[v["clusterVersion"]] = { "status": v.get("versionStatus") or v.get("status"), "fim_padrao": str(v.get("endOfStandardSupportDate") or ""), } if not resp.get("nextToken"): return status kw["nextToken"] = resp["nextToken"] except Exception as e: # noqa: BLE001 aviso(f"{regiao}: status das versões do EKS indisponível ({descrever_erro(e)}); conferir no console") return {} def coletar_eks(s, regiao, saida): eks = s.client("eks", region_name=regiao, config=CFG) nomes = list(paginar(eks, "list_clusters", "clusters")) if not nomes: return status_versoes = versoes_eks(eks, regiao) for nome in nomes: v = eks.describe_cluster(name=nome)["cluster"].get("version") saida.append({"tipo": "eks", "regiao": regiao, "id": nome, "versao": v, "suporte": status_versoes.get(v), "versoes_regiao": status_versoes}) def eh_al2(img): nome = (img.get("Name") or "").lower() desc = (img.get("Description") or "").lower() if "al2023" in nome or "2023" in desc: return False return nome.startswith("amzn2-") or nome.startswith("amazon-eks-node-1.") or "amazon linux 2 " in desc + " " def coletar_ec2_al2(s, regiao, saida): ec2 = s.client("ec2", region_name=regiao, config=CFG) instancias = [] for r in paginar(ec2, "describe_instances", "Reservations", Filters=[{"Name": "instance-state-name", "Values": ["running", "stopped"]}]): instancias += r.get("Instances", []) imagens = {} ids = sorted({i["ImageId"] for i in instancias if i.get("ImageId")}) for lote in (ids[i:i + 100] for i in range(0, len(ids), 100)): try: imagens.update({img["ImageId"]: img for img in ec2.describe_images(ImageIds=lote)["Images"]}) except ClientError: # alguma AMI apagada ou inválida derruba o lote: consulta uma a uma for iid in lote: try: imagens.update({img["ImageId"]: img for img in ec2.describe_images(ImageIds=[iid])["Images"]}) except ClientError: pass for i in instancias: img = imagens.get(i.get("ImageId")) if img is None: if "linux" in (i.get("PlatformDetails") or "").lower(): saida.append({"tipo": "ec2-indeterminado", "regiao": regiao, "id": i["InstanceId"], "ami": i.get("ImageId")}) continue if eh_al2(img): saida.append({"tipo": "ec2-al2", "regiao": regiao, "id": i["InstanceId"], "ami": i["ImageId"], "ami_nome": img.get("Name"), "classe": i.get("InstanceType")}) def custos_ja_pagos(s): """Últimos meses de cobrança com 'ExtendedSupport' no tipo de uso (Cost Explorer: só na conta pagadora).""" ce = s.client("ce", region_name="us-east-1", config=CFG) hoje = dt.date.today() inicio = (hoje.replace(day=1) - dt.timedelta(days=75)).replace(day=1) try: resp = ce.get_cost_and_usage(TimePeriod={"Start": inicio.isoformat(), "End": hoje.isoformat()}, Granularity="MONTHLY", Metrics=["UnblendedCost"], GroupBy=[{"Type": "DIMENSION", "Key": "USAGE_TYPE"}]) except Exception as e: # noqa: BLE001 aviso(f"Cost Explorer indisponível ({descrever_erro(e)}); o relatório sai sem o histórico de cobrança") return [] linhas = [] for periodo in resp.get("ResultsByTime", []): for g in periodo.get("Groups", []): tipo = g["Keys"][0] if "extendedsupport" in tipo.replace("-", "").replace("_", "").lower(): linhas.append({"mes": periodo["TimePeriod"]["Start"][:7], "tipo_uso": tipo, "usd": round(float(g["Metrics"]["UnblendedCost"]["Amount"]), 2)}) return linhas def anonimizar(dados): """Troca nomes e IDs por apelidos estáveis (banco-01, cache-01...) e devolve o mapa para o cliente guardar.""" apelidos, contagem = {}, defaultdict(int) def ap(espaco, real): if not real: return real if (espaco, real) not in apelidos: contagem[espaco] += 1 apelidos[(espaco, real)] = f"{espaco}-{contagem[espaco]:02d}" return apelidos[(espaco, real)] for r in dados["recursos"]: t = r["tipo"] if t == "rds": r["id"], r["cluster"], r["replica_de"] = ap("banco", r["id"]), ap("cluster", r.get("cluster")), ap("banco", r.get("replica_de")) elif t == "elasticache": r["id"], r["grupo"] = ap("cache-no", r["id"]), ap("cache", r["grupo"]) elif t == "opensearch": r["id"] = ap("busca", r["id"]) elif t == "eks": r["id"] = ap("eks", r["id"]) elif t.startswith("ec2"): r["id"], r["ami"] = ap("ec2", r["id"]), ap("ami", r.get("ami")) nome = r.get("ami_nome") or "" if not (nome.startswith("amzn2-") or nome.startswith("amazon-eks-node-")): r["ami_nome"] = None # nome de AMI própria pode revelar o sistema; as públicas da AWS ficam conta = dados["conta"] dados["conta"] = f"{conta[:4]}****{conta[-4:]}" dados["anonimizado"] = True return { "_leia": "Guarde este arquivo com você: ele liga os apelidos do relatório aos nomes reais. Não precisa enviar.", "conta": conta, "apelidos": {apelido: real for (_, real), apelido in apelidos.items()}, } def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--cliente", default="", help="nome da empresa (aparece no relatório)") ap.add_argument("--regioes", nargs="*", help="padrão: todas as regiões habilitadas na conta") ap.add_argument("--anonimizar", action="store_true", help="troca nomes e IDs dos recursos por apelidos") ap.add_argument("--sem-custos", action="store_true", help="não consulta o Cost Explorer") ap.add_argument("--saida", default="raio-x-coleta.json") ap.add_argument("--perfil", help=argparse.SUPPRESS) ap.add_argument("--role-arn", help=argparse.SUPPRESS) ap.add_argument("--external-id", help=argparse.SUPPRESS) a = ap.parse_args() sys.stdout.reconfigure(line_buffering=True) # mantém o progresso e os avisos na ordem certa caminho = os.path.abspath(a.saida) os.makedirs(os.path.dirname(caminho), exist_ok=True) print(f"Raio-X de suporte estendido · Observei · coletor {VERSAO}") s = sessao(a) try: conta = s.client("sts").get_caller_identity()["Account"] except Exception as e: # noqa: BLE001 sys.exit(f"Não consegui identificar a conta AWS ({descrever_erro(e)}). Rode no CloudShell, logado na conta.") regioes = a.regioes if not regioes: try: regioes = [r["RegionName"] for r in s.client("ec2", region_name="us-east-1", config=CFG).describe_regions()["Regions"]] except Exception as e: # noqa: BLE001 regioes = sorted({s.region_name or "sa-east-1", "sa-east-1", "us-east-1"}) aviso(f"lista de regiões indisponível ({descrever_erro(e)}); analisando só {', '.join(regioes)}") print(f"Conta {conta} · {len(regioes)} região(ões) · leva de 1 a 3 minutos") recursos = [] etapas = [("RDS/Aurora", coletar_rds), ("ElastiCache", coletar_elasticache), ("OpenSearch", coletar_opensearch), ("EKS", coletar_eks), ("EC2/Amazon Linux 2", coletar_ec2_al2)] for regiao in regioes: antes = len(recursos) for nome, fn in etapas: try: fn(s, regiao, recursos) except Exception as e: # noqa: BLE001 — uma etapa com problema não pode derrubar a coleta inteira aviso(f"{regiao}: {nome} não coletado ({descrever_erro(e)})") print(f"- {regiao}: {len(recursos) - antes} recurso(s)") dados = { "versao_coletor": VERSAO, "cliente": a.cliente, "conta": conta, "coletado_em": dt.datetime.now().isoformat(timespec="seconds"), "regioes": regioes, "recursos": recursos, "cobrancas_extended_support": [] if a.sem_custos else custos_ja_pagos(s), "avisos": avisos, "anonimizado": False, } if a.anonimizar: mapa = anonimizar(dados) caminho_mapa = os.path.join(os.path.dirname(caminho), "raio-x-mapa-local.json") with open(caminho_mapa, "w", encoding="utf-8") as f: json.dump(mapa, f, ensure_ascii=False, indent=2) print(f"Mapa de apelidos (fica com você): {caminho_mapa}") with open(caminho, "w", encoding="utf-8") as f: json.dump(dados, f, ensure_ascii=False, indent=2, default=str) print(f"\nPronto: {len(recursos)} recurso(s) e {len(avisos)} aviso(s) em {caminho}") print(f"No CloudShell, baixe pelo menu Ações (Actions) > Baixar arquivo (Download file) > {caminho}") print(f"Depois envie o arquivo para a Observei: {CONTATO}.") if __name__ == "__main__": main()