Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n_gcp/resources/gke.py: 62%

1# Copyright The Cloud Custodian Authors. 

2# SPDX-License-Identifier: Apache-2.0 

3import re 

4 

5from c7n_gcp.provider import resources 

6from c7n_gcp.query import (QueryResourceManager, TypeInfo, ChildTypeInfo, 

7 ChildResourceManager) 

8from c7n.utils import type_schema, local_session 

9from c7n_gcp.actions import MethodAction 

10from c7n_gcp.utils import get_firewall_port_ranges 

11 

12from c7n.filters import ValueFilter 

13 

14@resources.register('gke-cluster') 

15class KubernetesCluster(QueryResourceManager): 

16 """GCP resource: 

17 https://cloud.google.com/kubernetes-engine/docs/reference/rest/v1/projects.locations.clusters 

18 """ 

19 

20 class resource_type(TypeInfo): 

21 service = 'container' 

22 version = 'v1' 

23 component = 'projects.locations.clusters' 

24 enum_spec = ('list', 'clusters[]', None) 

25 scope = 'project' 

26 scope_key = 'parent' 

27 scope_template = "projects/{}/locations/-" 

28 name = id = "name" 

29 default_report_fields = [ 

30 'name', 'description', 'status', 'currentMasterVersion', 'currentNodeVersion', 

31 'currentNodeCount', 'location'] 

32 asset_type = 'container.googleapis.com/Cluster' 

33 scc_type = 'google.container.Cluster' 

34 metric_key = 'resource.labels.cluster_name' 

35 urn_component = 'cluster' 

36 labels = True 

37 labels_op = 'setResourceLabels' 

38 urn_zonal = True 

39 

40 @staticmethod 

41 def get(client, resource_info): 

42 return client.execute_query( 

43 'get', verb_arguments={ 

44 'name': 'projects/{}/locations/{}/clusters/{}'.format( 

45 resource_info['project_id'], 

46 resource_info['location'], 

47 resource_info['cluster_name'])}) 

48 

49 @staticmethod 

50 def get_label_params(resource, all_labels): 

51 location_str = "locations" 

52 if resource['selfLink'].find(location_str) < 0: 

53 location_str = "zones" 

54 path_param_re = re.compile('.*?/projects/(.*?)/'+location_str+'/(.*?)/clusters/(.*)') 

55 project, zone, cluster_name = path_param_re.match( 

56 resource['selfLink']).groups() 

57 return {'name': 'projects/'+project+'/locations/'+zone+'/clusters/'+cluster_name, 

58 'body': { 

59 'resourceLabels': all_labels, 

60 'labelFingerprint': resource['labelFingerprint'] 

61 }} 

62 

63 @classmethod 

64 def refresh(cls, client, resource): 

65 project_id = resource['selfLink'].split("/")[5] 

66 return cls.get( 

67 client, 

68 { 

69 'project_id': project_id, 

70 'location': resource['zone'], 

71 'cluster_name': resource['name'] 

72 } 

73 ) 

74 

75 def augment(self, resources): 

76 if not resources: 

77 return [] 

78 for r in resources: 

79 if r.get('resourceLabels'): 

80 r['labels'] = r['resourceLabels'] 

81 return resources 

82 

83@KubernetesCluster.filter_registry.register('effective-firewall') 

84class EffectiveFirewall(ValueFilter): 

85 """Filters gke clusters by their effective firewall rules. 

86 See `getEffectiveFirewalls 

87 https://cloud.google.com/workflows/docs/reference/googleapis/compute/v1/networks/getEffectiveFirewalls`_ 

88 for valid fields. 

89 

90 :example: 

91 

92 Filter all gke clusters that have a firewall rule that allows public 

93 access 

94 

95 .. code-block:: yaml 

96 

97 policies: 

98 - name: find-publicly-accessable-clusters 

99 resource: gcp.gke-cluster 

100 filters: 

101 - type: effective-firewall 

102 key: sourceRanges[] 

103 op: contains 

104 value: "0.0.0.0/0" 

105 """ 

106 

107 schema = type_schema('effective-firewall', rinherit=ValueFilter.schema) 

108 permissions = ('compute.instances.getEffectiveFirewalls',) 

109 annotation_key = "c7n:firewall" 

110 

111 def get_firewalls(self, client, p, r): 

112 if self.annotation_key not in r: 

113 firewalls = client.execute_command('getEffectiveFirewalls', 

114 verb_arguments = {'project': p, 'network': r['network']}).get('firewalls', []) 

115 

116 r[self.annotation_key] = get_firewall_port_ranges(firewalls) 

117 return super(EffectiveFirewall, self).process(r[self.annotation_key], None) 

118 

119 def process(self, resources, event=None): 

120 session = local_session(self.manager.session_factory) 

121 project = session.get_default_project() 

122 client = session.client( 

123 "compute", "v1", "networks" 

124 ) 

125 resource_list = [r for r in resources 

126 if self.get_firewalls(client, project, r)] 

127 return resource_list 

128 

129 

130@resources.register('gke-nodepool') 

131class KubernetesClusterNodePool(ChildResourceManager): 

132 """GCP resource: 

133 https://cloud.google.com/kubernetes-engine/docs/reference/rest/v1/projects.zones.clusters.nodePools 

134 """ 

135 

136 def _get_parent_resource_info(self, child_instance): 

137 project_param_re = re.compile( 

138 '.*?/projects/(.*?)/zones/(.*?)/clusters/(.*?)/nodePools/(.*?)' 

139 ) 

140 parent_values = re.match(project_param_re, child_instance['selfLink']).groups() 

141 parent_info = dict( 

142 zip(('project_id', 'location', 'cluster_name', 'node_name'), parent_values) 

143 ) 

144 

145 return parent_info 

146 

147 def _get_child_enum_args(self, parent_instance): 

148 return { 

149 'parent': 'projects/{}/locations/{}/clusters/{}'.format( 

150 local_session(self.session_factory).get_default_project(), 

151 parent_instance['location'], 

152 parent_instance['name'] 

153 ) 

154 } 

155 

156 class resource_type(ChildTypeInfo): 

157 service = 'container' 

158 version = 'v1' 

159 component = 'projects.locations.clusters.nodePools' 

160 enum_spec = ('list', 'nodePools[]', None) 

161 scope = 'global' 

162 name = id = 'name' 

163 parent_spec = {'resource': 'gke-cluster'} 

164 asset_type = 'container.googleapis.com/NodePool' 

165 default_report_fields = ['name', 'status', 'version'] 

166 permissions = ('container.nodes.list',) 

167 urn_component = 'cluster-node-pool' 

168 urn_zonal = True 

169 

170 @staticmethod 

171 def get(client, resource_info): 

172 cluster_name = resource_info['cluster_name'] 

173 name = re.match( 

174 r".*{}-(.*)-[^-]+-[^-]?".format(cluster_name), 

175 resource_info['resourceName']).group(1) 

176 

177 return client.execute_command( 

178 'get', verb_arguments={ 

179 'name': 'projects/{}/locations/{}/clusters/{}/nodePools/{}'.format( 

180 resource_info['project_id'], 

181 resource_info['location'], 

182 resource_info['cluster_name'], 

183 name)} 

184 ) 

185 

186 @classmethod 

187 def _get_location(cls, resource): 

188 "Get the region from the parent - the cluster" 

189 return super()._get_location(cls.get_parent(resource)) 

190 

191@KubernetesCluster.filter_registry.register('server-config') 

192@KubernetesClusterNodePool.filter_registry.register('server-config') 

193class ServerConfig(ValueFilter): 

194 """Filters kubernetes clusters or nodepools by their server config. 

195 See `getServerConfig 

196 https://cloud.google.com/kubernetes-engine/docs/reference/rest/v1/projects.locations/getServerConfig` 

197 for valid fields. 

198 

199 :example: 

200 

201 Filter all clusters that is not running a supported version 

202 

203 .. code-block:: yaml 

204 

205 policies: 

206 - name: find-unsupported-cluster-version 

207 resource: gcp.gke-cluster 

208 filters: 

209 - type: server-config 

210 key: contains(serverConfig.validMasterVersions, resource.currentMasterVersion) 

211 value: false 

212 

213 Filter all nodepools that is not running a supported version 

214 

215 .. code-block:: yaml 

216 

217 policies: 

218 - name: find-unsupported-cluster-nodepools-version 

219 resource: gcp.gke-nodepool 

220 filters: 

221 - type: server-config 

222 key: contains(serverConfig.validNodeVersions, resource.version) 

223 value: false 

224 """ 

225 

226 schema = type_schema('server-config', rinherit=ValueFilter.schema) 

227 permissions = ('container.nodes.get', 'container.clusters.get') 

228 annotation_key = "c7n:config" 

229 

230 def _get_location(self, r): 

231 return r["location"] if "location" in r else r['selfLink'].split('/')[-5] 

232 

233 def get_config(self, client, project, resource): 

234 if self.annotation_key in resource: 

235 return 

236 location = self._get_location(resource) 

237 resource[self.annotation_key] = client.execute_command( 

238 'getServerConfig', verb_arguments={ 

239 'name': 'projects/{}/locations/{}'.format(project, location)} 

240 ) 

241 

242 def __call__(self, r): 

243 return super().__call__({"serverConfig": r[self.annotation_key], "resource": r}) 

244 

245 def process(self, resources, event=None): 

246 session = local_session(self.manager.session_factory) 

247 project = session.get_default_project() 

248 client = session.client("container", "v1", "projects.locations") 

249 for r in resources: 

250 self.get_config(client, project, r) 

251 return super().process(resources) 

252 

253 

254@KubernetesCluster.action_registry.register('delete') 

255class Delete(MethodAction): 

256 """Action to delete GKE clusters 

257 

258 It is recommended to use a filter to avoid unwanted deletion of GKE clusters 

259 

260 :example: 

261 

262 .. code-block:: yaml 

263 

264 policies: 

265 - name: gcp-delete-testing-gke-clusters 

266 resource: gcp.gke-cluster 

267 filters: 

268 - type: value 

269 key: name 

270 op: regex 

271 value: '^(test-|demo-)*' 

272 actions: 

273 - type: delete 

274 """ 

275 

276 schema = type_schema('delete') 

277 method_spec = {'op': 'delete'} 

278 

279 def get_resource_params(self, model, resource_info): 

280 project = local_session(self.manager.source.query.session_factory).get_default_project() 

281 

282 return {'name': 'projects/{}/locations/{}/clusters/{}'.format( 

283 project, 

284 resource_info['location'], 

285 resource_info['name'])}