最近由于工作原因接触到aws的服务,我需要实时获取所有的域名信息,用于对其进行扫描,因此写了一个自动化爬取脚本 给需要的人分享。
1.基础准备
代码环境:python3
第三方库:boto3 (安装方法pip install boto3)
官方文档:https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/route53.html#route53
2.获取client
首先你需要获取一个有效的key,指路
控制台 -> IAM ->安全凭证 ->访问密钥,得到key后就可以正式开始编程了
#授权的key
access_key_id = ""
secret_access_key = ""
client = boto3.client('route53', aws_access_key_id=access_key_id,aws_secret_access_key=secret_access_key)
如果key的权限正常的话,这个client就能用来获取我们需要的dns数据了。
3.获取区域
route53的区域指的是根域名,每一个根域名都会有一个独立的区域,如果我们想要获取具体的解析记录,就需要先获取所有的域名id。
#获取账号下的区域id
def get_hostedzone_id(client):return client.list_hosted_zones(MaxItems='100',#Marker='',#DelegationSetId='string',#HostedZoneType='PrivateHostedZone')
- 单次查询的最大记录是100条
- marker,delegationsetid 是区域数目超过100时遍历查询是需要使用的参数,不超过100的情况下不需要
- HostedZoneType可以用来指定查public或private区域,不写就默认全部
- private区域的域名只能在aws内网中使用
正常的响应语法如下
{'HostedZones': [{'Id': 'string','Name': 'string','CallerReference': 'string','Config': {'Comment': 'string','PrivateZone': True|False},'ResourceRecordSetCount': 123,'LinkedService': {'ServicePrincipal': 'string','Description': 'string'}},],'Marker': 'string','IsTruncated': True|False,'NextMarker': 'string','MaxItems': 'string'
}
- IsTruncated = False 就表示查询数据已经到头了
4.获取指定区域的全部域名解析记录
#获取指定区域下的所有dns解析记录
def get_dns_records(client, hostedzone_id):#数据量低于300response = client.list_resource_record_sets(HostedZoneId=hostedzone_id,MaxItems='300')dns_records = response['ResourceRecordSets']#数据量超出300部分循环while(response['IsTruncated'] != False):response = client.list_resource_record_sets(HostedZoneId=hostedzone_id,StartRecordName=response["NextRecordName"],StartRecordType=response["NextRecordType"],MaxItems='300') dns_records.extend(response['ResourceRecordSets']) return dns_records
- 解析记录的单次查询上限是300条,超过的话就需要根据NextRecordName和NextRecordType 循环查询。
5.获取账号下的全部解析记录
通过上面2个函数组合一下,我们就能获取账号下的全部dns解析记录
def get_all_dns_records(client):dns_records = []hostedzones = get_hostedzone_id(client)for zone in hostedzones["HostedZones"]:dns_records.extend(get_dns_records(client, zone['Id']))return dns_records
正常的响应结果如下
{'ResourceRecordSets': [{'Name': 'string','Type': 'SOA'|'A'|'TXT'|'NS'|'CNAME'|'MX'|'NAPTR'|'PTR'|'SRV'|'SPF'|'AAAA'|'CAA'|'DS','SetIdentifier': 'string','Weight': 123,'Region': 'us-east-1'|'us-east-2'|'us-west-1'|'us-west-2'|'ca-central-1'|'eu-west-1'|'eu-west-2'|'eu-west-3'|'eu-central-1'|'eu-central-2'|'ap-southeast-1'|'ap-southeast-2'|'ap-southeast-3'|'ap-northeast-1'|'ap-northeast-2'|'ap-northeast-3'|'eu-north-1'|'sa-east-1'|'cn-north-1'|'cn-northwest-1'|'ap-east-1'|'me-south-1'|'me-central-1'|'ap-south-1'|'ap-south-2'|'af-south-1'|'eu-south-1'|'eu-south-2'|'ap-southeast-4'|'il-central-1','GeoLocation': {'ContinentCode': 'string','CountryCode': 'string','SubdivisionCode': 'string'},'Failover': 'PRIMARY'|'SECONDARY','MultiValueAnswer': True|False,'TTL': 123,'ResourceRecords': [{'Value': 'string'},],'AliasTarget': {'HostedZoneId': 'string','DNSName': 'string','EvaluateTargetHealth': True|False},'HealthCheckId': 'string','TrafficPolicyInstanceId': 'string','CidrRoutingConfig': {'CollectionId': 'string','LocationName': 'string'}},],'IsTruncated': True|False,'NextRecordName': 'string','NextRecordType': 'SOA'|'A'|'TXT'|'NS'|'CNAME'|'MX'|'NAPTR'|'PTR'|'SRV'|'SPF'|'AAAA'|'CAA'|'DS','NextRecordIdentifier': 'string','MaxItems': 'string'
}
- 实测 NextRecordIdentifier 并没有返回 ,也不影响查询结果
6.获取指定的DNS解析记录
record_type = ['A','AAAA',"CNAME"]
#根据想要的dns记录筛选最终数据
def get_dns_records_by_type(dns_records, record_type):final_dns_records =[]for record in dns_records:if record['Type'] in record_type:final_dns_records.append(record)return final_dns_records
- record_type 为需要提取的DNS解析类型,总共有 A | AAAA | CAA | CNAME | MX | NAPTR | NS | PTR | SOA | SPF | SRV | TXT 这些类型
- 这里的添加集合用的是append,之前用的都有extend 具体有啥区别 感兴趣的可以自行研究
7.去重 + 公网开放测试
由于部分cname本身并不是有效域名 只是一个单纯转发,且dns解析上无法判断解析记录是否公网开放,所以需要进行测试。这里提供一个简单的方案 就是直接发起request请求,如果有响应则证明解析有效。
#测试网站是否能访问,这里使用set进行去重
def test_web_alive(dns_records):web_list = []dns_list = set()for dns in dns_records:dns_list.add(dns['Name'][:-1])print(len(dns_list))for dns in dns_list:try:response = requests.get("https://"+dns)web_list.append([dns,response.status_code,'aws'])except:web_list.append([dns,'cant reach','aws'])return web_list #将测试结果存储到excel,默认第一行为表头
def save_dns_to_xlsx(web_list, path):workbook = openpyxl.load_workbook(path)sheet =workbook["Sheet1"] # 默认存到第一页for index,dns in enumerate(web_list):sheet.cell(row=index+2,column=1).value = dns[0]sheet.cell(row=index+2,column=2).value = dns[1]sheet.cell(row=index+2,column=3).value = dns[2]workbook.save(path)
如果有问题,欢迎留言咨询~