【发布时间】：2020-01-08 09:41:28
【问题描述】：
例如，我有一个名为test 的实例，现在我需要知道为这个实例打开了哪个端口。如何使用gcloud 在命令行中执行此操作？
不是实例内部打开的端口，而是附加到该实例的防火墙。
【问题讨论】：
标签： google-cloud-platform google-compute-engine
【发布时间】：2020-01-08 09:41:28
【问题描述】：
可以通过 2 个步骤查看与使用云 shell 的实例关联的防火墙规则。首先，我们必须运行以下命令来获取实例详细信息以查看防火墙标签：gcloud compute instances describe instance-name
在输出中，您将看到如下所示的防火墙标签：
然后运行以下命令查看这些标签附加到哪个防火墙规则。
gcloud compute firewall-rules list --format="table(name,network,
direction,priority,sourceRanges.list():label=SRC_RANGES,
destinationRanges.list():label=DEST_RANGES,
allowed[].map().firewall_rule().list():label=ALLOW,
denied[].map().firewall_rule().list():label=DENY,
sourceTags.list():label=SRC_TAGS,
sourceServiceAccounts.list():label=SRC_SVC_ACCT,
targetTags.list():label=TARGET_TAGS,
targetServiceAccounts.list():label=TARGET_SVC_ACCT,
disabled
)"
它会给出如下输出： output of above command
【讨论】：
Cloud SDK 没有针对此要求的命令。您可以使用 gcloud 列出防火墙规则，您可以使用 gcloud 列出计算实例。您必须应用外部逻辑将两者映射在一起。
防火墙规则通过几种方法与计算引擎实例相关联：
- 通过目标标签
- 通过服务帐号
- 对于网络中的所有实例
因此，首先显示所有计算引擎实例，获取每个服务帐户和标签。然后显示所有防火墙规则，获取每个规则的目标。然后将所有内容匹配在一起并打印一个实例或所有实例的开放端口列表。
这对 CLI 来说太复杂了。您需要编写一个程序来执行此操作，或者实现一个脚本来处理、排序和同步 CLI 输出。
【讨论】：
我意识到这个问题有点老了，但我想添加一个详细的答案，完全自动化你需要的东西，因为我也需要它。希望其他人会发现它有用。
如上所述，GCP 防火墙规则可以通过三种方式应用：
- 网络标签
- 服务帐号
- VPC 成员资格
可以通过两个不同的gcloud 命令提取此数据，但您必须将每个单独的计算实例与我上面提到的三个项目之间的点连接起来。
我实际上需要在工作中执行此操作，以便生成针对具有暴露于公共 Internet 的端口的实例的 nmap 脚本。所以我过滤了一点，只包含正在运行、具有公共 IP 地址并且与源范围为 0.0.0.0/0 的防火墙规则相关联的实例。
这是我的做法。
首先，我生成 JSON 文件，这些文件经过专门格式化并使用我需要的数据进行过滤。我使用这两个命令：
gcloud compute firewall-rules list \
--format="json(name,allowed[].map().firewall_rule().list(),network,
targetServiceAccounts.list(),targetTags.list())" \
--filter="direction:INGRESS AND disabled:False AND
sourceRanges.list():0.0.0.0/0 AND
allowed[].map().firewall_rule().list():*" \
| tee ./json-data/"$i"/firewall-rules.json
gcloud compute instances list \
--format="json(name,networkInterfaces[].accessConfigs[0].natIP,
serviceAccounts[].email,tags.items[],networkInterfaces[].network)" \
--filter="networkInterfaces[].accessConfigs[0].type:ONE_TO_ONE_NAT
AND status:running" \
| tee ./json-data/"$i"/compute-instances.json
然后，我使用这个 python 脚本来处理上面生成的文件。它将在您的 CWD 中创建一个名为 out-firewall-data 的目录，其中包含三个文件：
- applied-rules.csv (instance,external_ip,allowed_tcp,allowed_udp)
- run-nmap.sh（端口扫描脚本）
- run-masscan.sh（对于 1-65535 暴露的实例，扫描速度更快）
我将这个脚本托管在GitLab repo here 中，并且可能会在那里继续开发它。它还支持一次审核多个项目，您可以查看存储库中的说明。
你可以像gen_nmap.py --single /path/to/jsons/一样运行这个脚本
#!/usr/bin/env python3
"""
Process gcloud output to determine applied firewall rules.
Firewall rules are applied via multiple methods and Google does not provide
an easy way to script what rules are actually applied to what compute
instances.
Please see the included README for detailed instructions.
"""
import glob
import sys
import os
import json
import argparse
def process_args():
"""Handles user-passed parameters"""
parser = argparse.ArgumentParser()
target = parser.add_mutually_exclusive_group(required=True)
target.add_argument('--single', '-s', type=str, action='store',
help='Single directory containing json files.')
target.add_argument('--multi', '-m', type=str, action='store',
help='Root directory contains multiple subdirectories'
' of json files')
args = parser.parse_args()
if args.single:
target = os.path.abspath(args.single)
else:
target = os.path.abspath(args.multi)
# Before moving on, validate all the input data is present
if not os.path.isdir(target):
print("[!] That directory does not exist. Please try again.")
sys.exit()
return args
def parse_json(file):
"""
Loads the json data from a file into memory
"""
# If used in multi mode, there is a good chance we hit a lot of empty
# or missing files. We'll return empty data on those so the program can
# continue with the next directory.
if not os.path.isfile(file):
return {}
with open(file, 'r') as infile:
try:
data = json.load(infile)
except json.decoder.JSONDecodeError:
return {}
return data
def cleanup_rules(rules):
"""
Extracts details from firewall rules for easier processing
"""
clean_rules = []
for rule in rules:
name = rule['name']
udp_ports = []
tcp_ports = []
if 'all' in rule['allowed']:
tcp_ports = ['all']
udp_ports = ['all']
else:
for ports in rule['allowed']:
if 'tcp' in ports:
tcp_ports = [port.replace('tcp:', '') for port in ports.split(',')]
if 'udp' in ports:
udp_ports = [port.replace('udp:', '') for port in ports.split(',')]
# If a rule set has no target tags and no target svc account
# then it is applied at the VPC level, so we grab that here.
if 'targetServiceAccounts' not in rule and 'targetTags' not in rule:
network = rule['network']
# Otherwise, we are not interested in the network and can discard
# it so that future functions will not think rules are applied
# network-wide.
else:
network = ''
# Tags and target svc accounts may or may not exist
if 'targetTags' in rule:
net_tags = rule['targetTags'].split(',')
else:
net_tags = []
if 'targetServiceAccounts' in rule:
svc_tags = rule['targetServiceAccounts'].split(',')
else:
svc_tags = []
clean_rules.append({'name': name,
'tcp_ports': tcp_ports,
'udp_ports': udp_ports,
'net_tags': net_tags,
'svc_tags': svc_tags,
'network': network})
return clean_rules
def cleanup_instances(instances):
"""
Extracts details from instace data for easier processing
"""
clean_instances = []
for instance in instances:
# The following values should exist for each instance due to the
# gcloud filtering used.
name = instance['name']
networks = [interface['network'] for interface in instance['networkInterfaces']]
external_ip = instance['networkInterfaces'][0]['accessConfigs'][0]['natIP']
# The following values may or may not exist, it depends how the
# instance is configured.
if 'serviceAccounts' in instance:
svc_account = instance['serviceAccounts'][0]['email']
else:
svc_account = ''
if 'tags' in instance:
tags = instance['tags']['items']
else:
tags = []
clean_instances.append({'name': name,
'tags': tags,
'svc_account': svc_account,
'networks': networks,
'external_ip': external_ip})
return clean_instances
def merge_dict(applied_rules, rule, instance):
"""
Adds or updates final entries into dictionary
Using a discrete function as several functions update this dictionary, so
we need to check for the existence of a key and then decide to create or
update it.
"""
name = instance['name']
if name in applied_rules:
applied_rules[name]['allowed_tcp'].update(rule['tcp_ports'])
applied_rules[name]['allowed_udp'].update(rule['udp_ports'])
else:
applied_rules[name] = {'external_ip': instance['external_ip'],
'allowed_tcp': set(rule['tcp_ports']),
'allowed_udp': set(rule['udp_ports'])}
return applied_rules
def process_tagged_rules(applied_rules, rules, instances):
"""
Extracts effective firewall rules applied by network tags on instances
"""
for rule in rules:
for instance in instances:
for tag in rule['net_tags']:
if tag in instance['tags']:
applied_rules = merge_dict(applied_rules, rule, instance)
return applied_rules
def process_vpc_rules(applied_rules, rules, instances):
"""
Extracts effective firewall rules applied by VPC membership
"""
for rule in rules:
for instance in instances:
# In the cleaning function, we only applied a network tag if the
# rule is applied to the whole VPC. So a match means it applies.
if rule['network'] and rule['network'] in instance['networks']:
applied_rules = merge_dict(applied_rules, rule, instance)
return applied_rules
def process_svc_rules(applied_rules, rules, instances):
"""
Extracts effective firewall rules applied by service accounts
"""
for rule in rules:
if rule['svc_tags']:
for instance in instances:
if instance['svc_account'] in rule['svc_tags']:
applied_rules = merge_dict(applied_rules, rule, instance)
return applied_rules
def process_output(applied_rules):
"""
Takes the python dictionary format and output several useful files
"""
if not applied_rules:
print("[!] No publicly exposed ports, sorry!")
sys.exit()
print("[*] Processing output for {} instances with exposed ports"
.format(len(applied_rules)))
out_dir = 'out-firewall-data'
if not os.path.exists(out_dir):
os.makedirs(out_dir)
# First, write the raw data in CSV
with open(out_dir + '/applied-rules.csv', 'w') as outfile:
outfile.write("name,external_ip,allowed_tcp,allowed_udp\n")
for i in applied_rules:
outfile.write("{},{},{},{}\n"
.format(i,
applied_rules[i]['external_ip'],
applied_rules[i]['allowed_tcp'],
applied_rules[i]['allowed_udp'])
.replace("set()", ""))
# Next, make an nmap script
nmap_tcp = 'nmap --open -Pn -sV -oX {}-tcp.xml {} -p {}\n'
nmap_tcp_common = 'nmap --open -Pn -sV -oX {}-tcp.xml {}\n'
nmap_udp = 'sudo nmap --open -Pn -sU -sV -oX {}-udp.xml {} -p {}\n'
nmap_udp_common = 'sudo nmap --open -Pn -sU -sV -oX {}-udp.xml {} -F\n'
with open(out_dir + '/run-nmap.sh', 'w') as outfile:
for name in applied_rules:
external_ip = applied_rules[name]['external_ip']
# "All" as a rule will apply to both TCP and UDP. These get special
# nmap commands to do only the common ports (full range is too slow
# and will be handled with masscan commands.
if set(['all']) in applied_rules[name].values():
outfile.write("echo running common TCP/UDP scans against {}\n"
.format(name))
outfile.write(nmap_tcp_common.format(name, external_ip))
outfile.write(nmap_udp_common.format(name, external_ip))
else:
if applied_rules[name]['allowed_tcp']:
ports = ','.join(applied_rules[name]['allowed_tcp'])
outfile.write("echo running TCP scans against {}\n"
.format(name))
outfile.write(nmap_tcp.format(name, external_ip, ports))
if applied_rules[name]['allowed_udp']:
ports = ','.join(applied_rules[name]['allowed_udp'])
outfile.write("echo running UDP scans against {}\n"
.format(name))
outfile.write(nmap_udp.format(name, external_ip, ports))
# Now, write masscan script for machines with all TCP ports open
masscan = 'sudo masscan -p{} {} --rate=1000 --open-only -oX {}.xml\n'
with open(out_dir + '/run-masscan.sh', 'w') as outfile:
for name in applied_rules:
external_ip = applied_rules[name]['external_ip']
if set(['all']) in applied_rules[name].values():
outfile.write("echo running full masscan against {}\n"
.format(name))
outfile.write(masscan.format('1-65535', external_ip, name))
print("[+] Wrote some files to {}, enjoy!".format(out_dir))
def main():
"""
Main function to parse json files and write analyzed output
"""
args = process_args()
applied_rules = {}
rules = []
instances = []
# Functions below in a loop based on whether we are targeting json files
# in a single directory or a tree with multiple project subdirectories.
if args.multi:
targets = glob.glob(args.multi + '/*')
else:
targets = [args.single]
for target in targets:
rules = parse_json(target + '/firewall-rules.json')
instances = parse_json(target + '/compute-instances.json')
if not rules or not instances:
print("[!] No valid data in {}".format(target))
continue
# Clean the data up a bit
rules = cleanup_rules(rules)
print("[*] Processed {} firewall rules in {}"
.format(len(rules), target))
instances = cleanup_instances(instances)
print("[*] Processed {} instances in {}"
.format(len(instances), target))
# Connect the dots and build out the applied rules dictionary
applied_rules = process_tagged_rules(applied_rules, rules, instances)
applied_rules = process_vpc_rules(applied_rules, rules, instances)
applied_rules = process_svc_rules(applied_rules, rules, instances)
# Process data and create various output files
process_output(applied_rules)
if __name__ == '__main__':
main()
【讨论】：