Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n/actions/network.py: 22%
114 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:51 +0000
1# Copyright The Cloud Custodian Authors.
2# SPDX-License-Identifier: Apache-2.0
3import itertools
5from c7n.exceptions import PolicyExecutionError, PolicyValidationError
6from c7n import utils
7import jmespath
9from .core import Action
12class ModifyVpcSecurityGroupsAction(Action):
13 """Common action for modifying security groups on a vpc attached resources.
15 Security groups for add or remove can be specified via group id or
16 name. Group removal also supports symbolic names such as
17 'matched', 'network-location' or 'all'. 'matched' uses the
18 annotations/output of the 'security-group' filter
19 filter. 'network-location' uses the annotations of the
20 'network-location' interface filter for `SecurityGroupMismatch`.
22 Note a vpc attached resource requires at least one security group,
23 this action will use the sg specified in `isolation-group` to ensure
24 resources always have at least one security-group.
26 type: modify-security-groups
27 add: []
28 remove: [] | matched | network-location
29 isolation-group: sg-xyz
30 add-by-tag: {}
32 :example:
34 .. code-block:: yaml
36 policies:
37 - name: set-prod-security-groups
38 resource: ec2
39 filters:
40 - type: value
41 key: 'tag:env'
42 value: 'prod'
43 actions:
44 - type: modify-security-groups
45 add: prod-default-sg
46 remove:
47 - launch-wizard-1
48 - launch-wizard-2
49 add-by-tag:
50 key: environment
51 values:
52 - production
53 """
54 schema_alias = True
55 schema = {
56 'type': 'object',
57 'additionalProperties': False,
58 'properties': {
59 'type': {'enum': ['modify-security-groups']},
60 'add': {'oneOf': [
61 {'type': 'string'},
62 {'type': 'array', 'items': {
63 'type': 'string'}}]},
64 'remove': {'oneOf': [
65 {'type': 'array', 'items': {
66 'type': 'string'}},
67 {'enum': [
68 'matched', 'network-location', 'all',
69 {'type': 'string'}]}]},
70 'isolation-group': {'oneOf': [
71 {'type': 'string'},
72 {'type': 'array', 'items': {
73 'type': 'string'}}]},
74 'add-by-tag': {
75 'type': 'object',
76 'additionalProperties': False,
77 'properties': {
78 'key': {'type': 'string'},
79 'values': {'type': 'array', 'items': {'type': 'string'}}
80 },
81 'required': ['key', 'values']
82 }
83 },
84 'anyOf': [
85 {'required': ['isolation-group', 'remove', 'type']},
86 {'required': ['add', 'remove', 'type']},
87 {'required': ['add', 'type']},
88 {'required': ['add-by-tag', 'type']}]
89 }
91 SYMBOLIC_SGS = {'all', 'matched', 'network-location'}
93 sg_expr = None
94 vpc_expr = None
96 def validate(self):
97 sg_filter = self.manager.filter_registry.get('security-group')
98 if not sg_filter or not sg_filter.RelatedIdsExpression:
99 raise PolicyValidationError(self._format_error((
100 "policy:{policy} resource:{resource_type} does "
101 "not support {action_type} action")))
102 if self.get_action_group_names():
103 vpc_filter = self.manager.filter_registry.get('vpc')
104 if not vpc_filter or not vpc_filter.RelatedIdsExpression:
105 raise PolicyValidationError(self._format_error((
106 "policy:{policy} resource:{resource_type} does not support "
107 "security-group names only ids in action:{action_type}")))
108 self.vpc_expr = utils.jmespath_compile(vpc_filter.RelatedIdsExpression)
109 if self.sg_expr is None:
110 self.sg_expr = utils.jmespath_compile(
111 self.manager.filter_registry.get('security-group').RelatedIdsExpression)
112 if 'all' in self._get_array('remove') and not self._get_array('isolation-group'):
113 raise PolicyValidationError(self._format_error((
114 "policy:{policy} use of action:{action_type} with "
115 "remove: all requires specifying isolation-group")))
116 return self
118 def get_group_names(self, groups):
119 names = []
120 for g in groups:
121 if g.startswith('sg-'):
122 continue
123 elif g in self.SYMBOLIC_SGS:
124 continue
125 names.append(g)
126 return names
128 def get_action_group_names(self):
129 """Return all the security group names configured in this action."""
130 return self.get_group_names(
131 list(itertools.chain(
132 *[self._get_array('add'),
133 self._get_array('remove'),
134 self._get_array('isolation-group')])))
136 def _format_error(self, msg, **kw):
137 return msg.format(
138 policy=self.manager.ctx.policy.name,
139 resource_type=self.manager.type,
140 action_type=self.type,
141 **kw)
143 def _get_array(self, k):
144 v = self.data.get(k, [])
145 if isinstance(v, (str, bytes)):
146 return [v]
147 return v
149 def get_groups_by_names(self, names):
150 """Resolve security names to security groups resources."""
151 if not names:
152 return []
153 client = utils.local_session(
154 self.manager.session_factory).client('ec2')
155 sgs = self.manager.retry(
156 client.describe_security_groups,
157 Filters=[{
158 'Name': 'group-name', 'Values': names}]).get(
159 'SecurityGroups', [])
161 unresolved = set(names)
162 for s in sgs:
163 if s['GroupName'] in unresolved:
164 unresolved.remove(s['GroupName'])
166 if unresolved:
167 raise PolicyExecutionError(self._format_error(
168 "policy:{policy} security groups not found "
169 "requested: {names}, found: {groups}",
170 names=list(unresolved), groups=[g['GroupId'] for g in sgs]))
171 return sgs
173 def get_groups_by_tag(self, key, values):
174 """Get security groups that match tag values."""
175 client = utils.local_session(
176 self.manager.session_factory).client('ec2')
177 sgs = self.manager.retry(
178 client.describe_security_groups,
179 Filters=[{
180 'Name': 'tag:' + key, 'Values': values}]).get(
181 'SecurityGroups', [])
182 return sgs
184 def resolve_group_names(self, r, target_group_ids, groups):
185 """Resolve any security group names to the corresponding group ids
187 With the context of a given network attached resource.
188 """
189 names = self.get_group_names(target_group_ids)
190 if not names:
191 return target_group_ids
193 target_group_ids = list(target_group_ids)
194 vpc_id = self.vpc_expr.search(r)
195 if not vpc_id:
196 raise PolicyExecutionError(self._format_error(
197 "policy:{policy} non vpc attached resource used "
198 "with modify-security-group: {resource_id}",
199 resource_id=r[self.manager.resource_type.id]))
201 found = False
202 for n in names:
203 for g in groups:
204 if g['GroupName'] == n and g['VpcId'] == vpc_id:
205 found = g['GroupId']
206 if not found:
207 raise PolicyExecutionError(self._format_error((
208 "policy:{policy} could not resolve sg:{name} for "
209 "resource:{resource_id} in vpc:{vpc}"),
210 name=n,
211 resource_id=r[self.manager.resource_type.id], vpc=vpc_id))
212 target_group_ids.remove(n)
213 target_group_ids.append(found)
214 return target_group_ids
216 def resolve_remove_symbols(self, r, target_group_ids, rgroups):
217 """Resolve the resources security groups that need be modified.
219 Specifically handles symbolic names that match annotations from policy filters
220 for groups being removed.
221 """
222 if 'matched' in target_group_ids:
223 return r.get('c7n:matched-security-groups', ())
224 elif 'network-location' in target_group_ids:
225 for reason in r.get('c7n:NetworkLocation', ()):
226 if reason['reason'] == 'SecurityGroupMismatch':
227 return list(reason['security-groups'])
228 elif 'all' in target_group_ids:
229 return rgroups
230 return target_group_ids
232 def get_groups(self, resources):
233 """Return lists of security groups to set on each resource
235 For each input resource, parse the various add/remove/isolation-
236 group/add-by-tag policies for 'modify-security-groups' to find the
237 resulting set of VPC security groups to attach to that resource.
239 Returns a list of lists containing the resulting VPC security groups
240 that should end up on each resource passed in.
242 :param resources: List of resources containing VPC Security Groups
243 :return: List of lists of security groups per resource
245 """
246 resolved_groups = self.get_groups_by_names(self.get_action_group_names())
247 return_groups = []
249 tag = self._get_array('add-by-tag')
250 if tag:
251 tag_filtered_groups = self.get_groups_by_tag(tag['key'], tag['values'])
252 else:
253 tag_filtered_groups = []
255 for idx, r in enumerate(resources):
256 rgroups = self.sg_expr.search(r) or []
257 add_groups = self.resolve_group_names(
258 r, self._get_array('add'), resolved_groups)
259 remove_groups = self.resolve_remove_symbols(
260 r,
261 self.resolve_group_names(
262 r, self._get_array('remove'), resolved_groups),
263 rgroups)
264 isolation_groups = self.resolve_group_names(
265 r, self._get_array('isolation-group'), resolved_groups)
267 for sg in tag_filtered_groups:
268 if sg['VpcId'] == jmespath.search(
269 self.manager.filter_registry.get('vpc').RelatedIdsExpression, r):
270 add_groups.append(sg['GroupId'])
272 for g in remove_groups:
273 if g in rgroups:
274 rgroups.remove(g)
275 for g in add_groups:
276 if g not in rgroups:
277 rgroups.append(g)
279 if not rgroups:
280 rgroups = list(isolation_groups)
282 if len(rgroups) > 5:
283 raise PolicyExecutionError(self._format_error(
284 "policy:{policy} - the number of security groups exceeds 5. "
285 "groups: {rgroups}",
286 rgroups=rgroups))
288 return_groups.append(rgroups)
290 return return_groups