Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/c7n/utils.py: 8%

1# Copyright The Cloud Custodian Authors.

2# SPDX-License-Identifier: Apache-2.0

3import copy

4from collections import UserString

5from datetime import datetime, timedelta

6from dateutil.tz import tzutc

7import json

8import itertools

9import ipaddress

10import logging

11import os

12import random

13import re

14import sys

15import threading

16import time

17import typing

18from urllib import parse as urlparse

19from urllib.request import getproxies, proxy_bypass

21from dateutil.parser import ParserError, parse

23import importlib.resources

24import jmespath

25from jmespath import functions

26from jmespath.parser import Parser, ParsedResult

28from c7n import config

29from c7n.exceptions import ClientError, PolicyValidationError

31# Try to play nice in a serverless environment, where we don't require yaml

33try:

34 import yaml

35except ImportError: # pragma: no cover

36 SafeLoader = BaseSafeDumper = yaml = None

37else:

38 try:

39 from yaml import CSafeLoader as SafeLoader, CSafeDumper as BaseSafeDumper

40 except ImportError: # pragma: no cover

41 from yaml import SafeLoader, SafeDumper as BaseSafeDumper

44class SafeDumper(BaseSafeDumper or object):

45 def ignore_aliases(self, data):

46 return True

49log = logging.getLogger('custodian.utils')

52class VarsSubstitutionError(Exception):

53 pass

56def load_file(path, format=None, vars=None):

57 if format is None:

58 format = 'yaml'

59 _, ext = os.path.splitext(path)

60 if ext[1:] == 'json':

61 format = 'json'

63 with open(path) as fh:

64 contents = fh.read()

66 if vars:

67 try:

68 contents = contents.format(**vars)

69 except IndexError:

70 msg = 'Failed to substitute variable by positional argument.'

71 raise VarsSubstitutionError(msg)

72 except KeyError as e:

73 msg = 'Failed to substitute variables. KeyError on {}'.format(str(e))

74 raise VarsSubstitutionError(msg)

76 if format == 'yaml':

77 return yaml_load(contents)

78 elif format == 'json':

79 return loads(contents)

82def yaml_load(value):

83 if yaml is None:

84 raise RuntimeError("Yaml not available")

85 return yaml.load(value, Loader=SafeLoader)

88def yaml_dump(value):

89 if yaml is None:

90 raise RuntimeError("Yaml not available")

91 return yaml.dump(value, default_flow_style=False, Dumper=SafeDumper)

94def loads(body):

95 return json.loads(body)

98def dumps(data, fh=None, indent=0):

99 if fh:

100 return json.dump(data, fh, cls=JsonEncoder, indent=indent)

101 else:

102 return json.dumps(data, cls=JsonEncoder, indent=indent)

103

104

105def format_event(evt):

106 return json.dumps(evt, indent=2)

107

108

109def filter_empty(d):

110 for k, v in list(d.items()):

111 if not v:

112 del d[k]

113 return d

114

115

116# We need a minimum floor when examining possible timestamp

117# values to distinguish from other numeric time usages. Use

118# the S3 Launch Date.

119DATE_FLOOR = time.mktime((2006, 3, 19, 0, 0, 0, 0, 0, 0))

120

121

122def parse_date(v, tz=None):

123 """Handle various permutations of a datetime serialization

124 to a datetime with the given timezone.

125

126 Handles strings, seconds since epoch, and milliseconds since epoch.

127 """

128

129 if v is None:

130 return v

131

132 tz = tz or tzutc()

133

134 if isinstance(v, datetime):

135 if v.tzinfo is None:

136 return v.astimezone(tz)

137 return v

138

139 if isinstance(v, str) and not v.isdigit():

140 try:

141 return parse(v).astimezone(tz)

142 except (AttributeError, TypeError, ValueError, OverflowError):

143 pass

144

145 # OSError on windows -- https://bugs.python.org/issue36439

146 exceptions = (ValueError, OSError) if os.name == "nt" else (ValueError)

147

148 if isinstance(v, (int, float, str)):

149 try:

150 if float(v) > DATE_FLOOR:

151 v = datetime.fromtimestamp(float(v)).astimezone(tz)

152 except exceptions:

153 pass

154

155 if isinstance(v, (int, float, str)):

156 # try interpreting as milliseconds epoch

157 try:

158 if float(v) > DATE_FLOOR:

159 v = datetime.fromtimestamp(float(v) / 1000).astimezone(tz)

160 except exceptions:

161 pass

162

163 return isinstance(v, datetime) and v or None

164

165

166# ISO 8601 durations built from fixed-length units. Years and months

167# vary in length, so a duration using them does not match.

168ISO_DURATION = re.compile(r"""

169 P

170 (?:(?P<weeks>\d+(?:[.,]\d+)?)W)?

171 (?:(?P<days>\d+(?:[.,]\d+)?)D)?

172 (?:T(?=\d)

173 (?:(?P<hours>\d+(?:[.,]\d+)?)H)?

174 (?:(?P<minutes>\d+(?:[.,]\d+)?)M)?

175 (?:(?P<seconds>\d+(?:[.,]\d+)?)S)?

176 )?

177""", re.VERBOSE | re.ASCII)

178

179

180def parse_duration(v):

181 """Parse an ISO 8601 duration such as ``PT30M`` or ``P1DT12H`` to a timedelta.

182

183 Weeks stand alone, only the smallest unit may have a fraction, and either

184 ``.`` or ``,`` marks the decimal. Returns None for anything else, including

185 durations in years or months and durations too large for a timedelta.

186 """

187 if not isinstance(v, str):

188 return None

189 match = ISO_DURATION.fullmatch(v.strip())

190 if match is None:

191 return None

192 parts = {unit: n for unit, n in match.groupdict().items() if n}

193 if not parts:

194 return None

195 # ISO 8601: weeks stand alone, and only the smallest unit given may have a fraction

196 if 'weeks' in parts and len(parts) > 1:

197 return None

198 if any(sep in n for n in list(parts.values())[:-1] for sep in '.,'):

199 return None

200 try:

201 return timedelta(**{unit: float(n.replace(',', '.')) for unit, n in parts.items()})

202 except OverflowError:

203 return None

204

205

206if typing.TYPE_CHECKING:

207 from c7n.element import ElementJSONSchema

208

209

210def type_schema(

211 type_name: str,

212 inherits: list[str] | None = None,

213 rinherit: 'ElementJSONSchema | None' = None,

214 aliases: list[str] | None = None,

215 required: list[str] | None = None,

216 **props: typing.Any) -> 'ElementJSONSchema':

217 """jsonschema generation helper

218

219 params:

220 - type_name: name of the type

221 - inherits: list of document fragments that are required via anyOf[$ref]

222 - rinherit: use another schema as a base for this, basically work around

223 inherits issues with additionalProperties and type enums.

224 - aliases: additional names this type maybe called

225 - required: list of required properties, by default 'type' is required

226 - props: additional key value properties

227 """

228 if aliases:

229 type_names = [type_name]

230 type_names.extend(aliases)

231 else:

232 type_names = [type_name]

233

234 if rinherit:

235 s = copy.deepcopy(rinherit)

236 s['properties']['type'] = {'enum': type_names}

237 else:

238 s = {

239 'type': 'object',

240 'properties': {

241 'type': {'enum': type_names}}}

242

243 # Ref based inheritance and additional properties don't mix well.

244 # https://stackoverflow.com/questions/22689900/json-schema-allof-with-additionalproperties

245 if not inherits:

246 s['additionalProperties'] = False

247

248 s['properties'].update(props)

249

250 for k, v in props.items():

251 if v is None:

252 del s['properties'][k]

253 if not required:

254 required = []

255 if isinstance(required, list):

256 required.append('type')

257 s['required'] = required

258 if inherits:

259 extended = s

260 s = {'allOf': [{'$ref': i} for i in inherits]}

261 s['allOf'].append(extended)

262 return s

263

264

265class JsonEncoder(json.JSONEncoder):

266

267 def default(self, obj):

268 if isinstance(obj, datetime):

269 return obj.isoformat()

270 if isinstance(obj, FormatDate):

271 return obj.datetime.isoformat()

272 if isinstance(obj, bytes):

273 return obj.decode('utf8', errors="ignore")

274 return json.JSONEncoder.default(self, obj)

275

276

277def group_by(resources, key):

278 """Return a mapping of key value to resources with the corresponding value.

279

280 Key may be specified as dotted form for nested dictionary lookup

281 """

282 resource_map = {}

283 parts = key.split('.')

284 for r in resources:

285 v = r

286 for k in parts:

287 v = v.get(k)

288 if not isinstance(v, dict):

289 break

290 resource_map.setdefault(v, []).append(r)

291 return resource_map

292

293

294def chunks(iterable, size=50):

295 """Break an iterable into lists of size"""

296 batch = []

297 for n in iterable:

298 batch.append(n)

299 if len(batch) % size == 0:

300 yield batch

301 batch = []

302 if batch:

303 yield batch

304

305

306def camelResource(obj, implicitDate=False, implicitTitle=True):

307 """Some sources from apis return lowerCased where as describe calls

308

309 always return TitleCase, this function turns the former to the later

310

311 implicitDate ~ automatically sniff keys that look like isoformat date strings

312 and convert to python datetime objects.

313 """

314 if not isinstance(obj, dict):

315 return obj

316 for k in list(obj.keys()):

317 v = obj.pop(k)

318 if implicitTitle:

319 ok = "%s%s" % (k[0].upper(), k[1:])

320 else:

321 ok = k

322 obj[ok] = v

323

324 if implicitDate:

325 # config service handles datetime differently then describe sdks

326 # the sdks use knowledge of the shape to support language native

327 # date times, while config just turns everything into a serialized

328 # json with mangled keys without type info. to normalize to describe

329 # we implicitly sniff keys which look like datetimes, and have an

330 # isoformat marker ('T').

331 kn = k.lower()

332 if isinstance(v, (str, int)) and ('time' in kn or 'date' in kn):

333 try:

334 dv = parse_date(v)

335 except ParserError:

336 dv = None

337 if dv:

338 obj[ok] = dv

339 if isinstance(v, dict):

340 camelResource(v, implicitDate, implicitTitle)

341 elif isinstance(v, list):

342 for e in v:

343 camelResource(e, implicitDate, implicitTitle)

344 return obj

345

346

347def get_account_id_from_sts(session):

348 response = session.client('sts').get_caller_identity()

349 return response.get('Account')

350

351

352def get_account_alias_from_sts(session):

353 response = session.client('iam').list_account_aliases()

354 aliases = response.get('AccountAliases', ())

355 return aliases and aliases[0] or ''

356

357

358def query_instances(session, client=None, **query):

359 """Return a list of ec2 instances for the query.

360 """

361 if client is None:

362 client = session.client('ec2')

363 p = client.get_paginator('describe_instances')

364 results = p.paginate(**query)

365 return list(itertools.chain(

366 *[r["Instances"] for r in itertools.chain(

367 *[pp['Reservations'] for pp in results])]))

368

369

370CONN_CACHE = threading.local()

371

372

373def local_session(factory, region=None):

374 """Cache a session thread local for up to 45m"""

375 factory_region = getattr(factory, 'region', 'global')

376 if region:

377 factory_region = region

378 s = getattr(CONN_CACHE, factory_region, {}).get('session')

379 t = getattr(CONN_CACHE, factory_region, {}).get('time')

380

381 n = time.time()

382 if s is not None and t + (60 * 45) > n:

383 return s

384 s = factory()

385

386 setattr(CONN_CACHE, factory_region, {'session': s, 'time': n})

387 return s

388

389

390def reset_session_cache():

391 for k in [k for k in dir(CONN_CACHE) if not k.startswith('_')]:

392 setattr(CONN_CACHE, k, {})

393

394 from .credentials import CustodianSession

395 CustodianSession.close()

396

397

398def annotation(i, k):

399 return i.get(k, ())

400

401

402def set_annotation(i, k, v):

403 """

404 >>> x = {}

405 >>> set_annotation(x, 'marker', 'a')

406 >>> annotation(x, 'marker')

407 ['a']

408 """

409 if not isinstance(i, dict):

410 raise ValueError("Can only annotate dictionaries")

411

412 if not isinstance(v, list):

413 v = [v]

414

415 if k in i:

416 ev = i.get(k)

417 if isinstance(ev, list):

418 ev.extend(v)

419 else:

420 i[k] = v

421

422

423def parse_s3(s3_path):

424 if not s3_path.startswith('s3://'):

425 raise ValueError("invalid s3 path")

426 ridx = s3_path.find('/', 5)

427 if ridx == -1:

428 ridx = None

429 bucket = s3_path[5:ridx]

430 s3_path = s3_path.rstrip('/')

431 if ridx is None:

432 key_prefix = ""

433 else:

434 key_prefix = s3_path[s3_path.find('/', 5):]

435 return s3_path, bucket, key_prefix

436

437

438REGION_PARTITION_MAP = json.loads(

439 (importlib.resources.files('c7n') / 'data/aws_region_partition_map.json').read_text()

440)

441

442

443def get_partition(region):

444 if not region:

445 return ''

446 return REGION_PARTITION_MAP.get(region, 'aws')

447

448

449def generate_arn(

450 service, resource, partition='aws',

451 region=None, account_id=None, resource_type=None, separator='/'):

452 """Generate an Amazon Resource Name.

453 See http://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html.

454 """

455 if region:

456 partition = get_partition(region)

457 if service == 's3':

458 region = ''

459 arn = 'arn:%s:%s:%s:%s:' % (

460 partition, service, region if region else '', account_id if account_id else '')

461 if resource_type:

462 if resource.startswith(separator):

463 separator = ''

464 arn = arn + '%s%s%s' % (resource_type, separator, resource)

465 else:

466 arn = arn + resource

467 return arn

468

469

470def snapshot_identifier(prefix, db_identifier):

471 """Return an identifier for a snapshot of a database or cluster.

472 """

473 now = datetime.now()

474 return '%s-%s-%s' % (prefix, db_identifier, now.strftime('%Y-%m-%d-%H-%M'))

475

476

477retry_log = logging.getLogger('c7n.retry')

478

479

480def get_retry(retry_codes=(), max_attempts=8, min_delay=1, log_retries=False):

481 """Decorator for retry boto3 api call on transient errors.

482

483 https://www.awsarchitectureblog.com/2015/03/backoff.html

484 https://en.wikipedia.org/wiki/Exponential_backoff

485

486 :param codes: A sequence of retryable error codes.

487 :param max_attempts: The max number of retries, by default the delay

488 time is proportional to the max number of attempts.

489 :param log_retries: Whether we should log retries, if specified

490 specifies the level at which the retry should be logged.

491 :param _max_delay: The maximum delay for any retry interval *note*

492 this parameter is only exposed for unit testing, as its

493 derived from the number of attempts.

494

495 Returns a function for invoking aws client calls that

496 retries on retryable error codes.

497 """

498 max_delay = max(min_delay, 2) ** max_attempts

499

500 def _retry(func, *args, ignore_err_codes=(), **kw):

501 for idx, delay in enumerate(

502 backoff_delays(min_delay, max_delay, jitter=True)):

503 try:

504 return func(*args, **kw)

505 except ClientError as e:

506 if e.response['Error']['Code'] in ignore_err_codes:

507 return

508 elif e.response['Error']['Code'] not in retry_codes:

509 raise

510 elif idx == max_attempts - 1:

511 raise

512 if log_retries:

513 retry_log.log(

514 log_retries,

515 "retrying %s on error:%s attempt:%d last delay:%0.2f",

516 func, e.response['Error']['Code'], idx, delay)

517 time.sleep(delay)

518 return _retry

519

520

521def backoff_delays(start, stop, factor=2.0, jitter=False):

522 """Geometric backoff sequence w/ jitter

523 """

524 cur = start

525 while cur <= stop:

526 if jitter:

527 yield cur - (cur * random.random() / 5)

528 else:

529 yield cur

530 cur = cur * factor

531

532

533def parse_cidr(value):

534 """Process cidr ranges."""

535 if isinstance(value, list) or isinstance(value, set):

536 return IPv4List([parse_cidr(item) for item in value])

537 klass = IPv4Network

538 if '/' not in value:

539 klass = ipaddress.ip_address

540 try:

541 v = klass(str(value))

542 except (ipaddress.AddressValueError, ValueError):

543 v = None

544 return v

545

546

547class IPv4Network(ipaddress.IPv4Network):

548

549 # Override for net 2 net containment comparison

550 def __contains__(self, other):

551 if other is None:

552 return False

553 if isinstance(other, ipaddress._BaseNetwork):

554 return self.supernet_of(other)

555 return super(IPv4Network, self).__contains__(other)

556

557 if (sys.version_info.major == 3 and sys.version_info.minor <= 6): # pragma: no cover

558 @staticmethod

559 def _is_subnet_of(a, b):

560 try:

561 # Always false if one is v4 and the other is v6.

562 if a._version != b._version:

563 raise TypeError(f"{a} and {b} are not of the same version")

564 return (b.network_address <= a.network_address and

565 b.broadcast_address >= a.broadcast_address)

566 except AttributeError:

567 raise TypeError(f"Unable to test subnet containment "

568 f"between {a} and {b}")

569

570 def supernet_of(self, other):

571 """Return True if this network is a supernet of other."""

572 return self._is_subnet_of(other, self)

573

574

575class IPv4List:

576 def __init__(self, ipv4_list):

577 self.ipv4_list = ipv4_list

578

579 def __contains__(self, other):

580 if other is None:

581 return False

582 in_networks = any([other in y_elem for y_elem in self.ipv4_list

583 if isinstance(y_elem, IPv4Network)])

584 in_addresses = any([other == y_elem for y_elem in self.ipv4_list

585 if isinstance(y_elem, ipaddress.IPv4Address)])

586 return any([in_networks, in_addresses])

587

588

589def reformat_schema(model):

590 """ Reformat schema to be in a more displayable format. """

591 if not hasattr(model, 'schema'):

592 return "Model '{}' does not have a schema".format(model)

593

594 if 'properties' not in model.schema:

595 return "Schema in unexpected format."

596

597 ret = copy.deepcopy(model.schema['properties'])

598

599 if 'type' in ret:

600 del ret['type']

601

602 for key in model.schema.get('required', []):

603 if key in ret:

604 ret[key]['required'] = True

605

606 return ret

607

608

609# from botocore.utils avoiding runtime dependency for botocore for other providers.

610# license apache 2.0

611def set_value_from_jmespath(source, expression, value, is_first=True):

612 # This takes a (limited) jmespath-like expression & can set a value based

613 # on it.

614 # Limitations:

615 # * Only handles dotted lookups

616 # * No offsets/wildcards/slices/etc.

617 bits = expression.split('.', 1)

618 current_key, remainder = bits[0], bits[1] if len(bits) > 1 else ''

619

620 if not current_key:

621 raise ValueError(expression)

622

623 if remainder:

624 if current_key not in source:

625 # We've got something in the expression that's not present in the

626 # source (new key). If there's any more bits, we'll set the key

627 # with an empty dictionary.

628 source[current_key] = {}

629

630 return set_value_from_jmespath(

631 source[current_key],

632 remainder,

633 value,

634 is_first=False

635 )

636

637 # If we're down to a single key, set it.

638 source[current_key] = value

639

640

641def format_string_values(obj, err_fallback=(IndexError, KeyError), formatter=None, *args, **kwargs):

642 """

643 Format all string values in an object.

644 Return the updated object

645 """

646 if isinstance(obj, dict):

647 new = {}

648 for key in obj.keys():

649 new[key] = format_string_values(obj[key], formatter=formatter, *args, **kwargs)

650 return new

651 elif isinstance(obj, list):

652 new = []

653 for item in obj:

654 new.append(format_string_values(item, formatter=formatter, *args, **kwargs))

655 return new

656 elif isinstance(obj, str):

657 try:

658 if formatter:

659 return formatter(obj, *args, **kwargs)

660 else:

661 return obj.format(*args, **kwargs)

662 except err_fallback:

663 return obj

664 else:

665 return obj

666

667

668def parse_url_config(url):

669 if url and '://' not in url:

670 url += "://"

671 conf = config.Bag()

672 parsed = urlparse.urlparse(url)

673 for k in ('scheme', 'netloc', 'path'):

674 conf[k] = getattr(parsed, k)

675 for k, v in urlparse.parse_qs(parsed.query).items():

676 conf[k] = v[0]

677 conf['url'] = url

678 return conf

679

680

681def join_output_path(output_path, *parts):

682 # allow users to specify interpolated output paths

683 if '{' in output_path:

684 return output_path

685

686 if "://" not in output_path:

687 return os.path.join(output_path, *parts)

688

689 # handle urls with query strings

690 parsed = urlparse.urlparse(output_path)

691 updated_path = "/".join((parsed.path, *parts))

692 parts = list(parsed)

693 parts[2] = updated_path

694 return urlparse.urlunparse(parts)

695

696

697def get_policy_provider(policy_data):

698 if isinstance(policy_data['resource'], list):

699 provider_name, _ = policy_data['resource'][0].split('.', 1)

700 elif '.' in policy_data['resource']:

701 provider_name, _ = policy_data['resource'].split('.', 1)

702 else:

703 provider_name = 'aws'

704 return provider_name

705

706

707def get_proxy_url(url):

708 proxies = getproxies()

709 parsed = urlparse.urlparse(url)

710

711 proxy_keys = [

712 parsed.scheme + '://' + parsed.netloc,

713 parsed.scheme,

714 'all://' + parsed.netloc,

715 'all'

716 ]

717

718 # Set port if not defined explicitly in url.

719 port = parsed.port

720 if port is None and parsed.scheme == 'http':

721 port = 80

722 elif port is None and parsed.scheme == 'https':

723 port = 443

724

725 hostname = parsed.hostname is not None and parsed.hostname or ''

726

727 # Determine if proxy should be used based on no_proxy entries.

728 # Note this does not support no_proxy ip or cidr entries.

729 if proxy_bypass("%s:%s" % (hostname, port)):

730 return None

731

732 for key in proxy_keys:

733 if key in proxies:

734 return proxies[key]

735

736 return None

737

738

739class DeferredFormatString(UserString):

740 """A string that returns itself when formatted

741

742 Let any format spec pass through. This lets us selectively defer

743 expansion of runtime variables without losing format spec details.

744 """

745 def __format__(self, format_spec):

746 return "".join(("{", self.data, f":{format_spec}" if format_spec else "", "}"))

747

748

749class FormatDate:

750 """a datetime wrapper with extended pyformat syntax"""

751

752 date_increment = re.compile(r'\+[0-9]+[Mdh]')

753

754 def __init__(self, d=None):

755 self._d = d

756

757 def __str__(self):

758 return str(self._d)

759

760 @property

761 def datetime(self):

762 return self._d

763

764 @classmethod

765 def utcnow(cls):

766 return cls(datetime.utcnow())

767

768 def __getattr__(self, k):

769 return getattr(self._d, k)

770

771 def __format__(self, fmt=None):

772 d = self._d

773 increments = self.date_increment.findall(fmt)

774 for i in increments:

775 p = {}

776 if i[-1] == 'M':

777 p['minutes'] = float(i[1:-1])

778 if i[-1] == 'h':

779 p['hours'] = float(i[1:-1])

780 if i[-1] == 'd':

781 p['days'] = float(i[1:-1])

782 d = d + timedelta(**p)

783 if increments:

784 fmt = self.date_increment.sub("", fmt)

785 return d.__format__(fmt)

786

787

788class QueryParser:

789

790 QuerySchema = {}

791 type_name = ''

792 # Allow multiple values to be passed to a query param

793 multi_value = True

794 # If using multi_value, specify scalar fields here

795 single_value_fields = ()

796

797 @classmethod

798 def is_implicit_query_filter(cls, data):

799 key = list(data[0].keys())[0]

800 if (key not in cls.QuerySchema and 'Filters' in cls.QuerySchema and

801 (key in cls.QuerySchema['Filters'] or key.startswith('tag:'))):

802 return True

803 return False

804

805 @classmethod

806 def implicit_qfilter_translate(cls, data):

807 filters = []

808 for d in data:

809 key = list(d.keys())[0]

810 values = list(d.values())[0]

811 if not isinstance(values, list):

812 values = [values]

813 filters.append({'Name': key, 'Values': values})

814 return [{'Filters': filters}]

815

816 @classmethod

817 def parse(cls, data):

818 if not isinstance(data, (tuple, list)):

819 raise PolicyValidationError(

820 f"{cls.type_name} Query Invalid Format, must be array of dicts"

821 )

822

823 # Backwards compatibility

824 if data:

825 if not isinstance(data[0], dict):

826 raise PolicyValidationError(

827 f"{cls.type_name} Query Invalid Format, must be array of dicts"

828 )

829 # Check for query filter key value pairs not listed under 'Filters' key

830 if cls.is_implicit_query_filter(data):

831 data = cls.implicit_qfilter_translate(data)

832

833 # Support iam-policy and elasticache 'Name', 'Value' queries without 'Filters' key

834 if (data[0].get('Value') and

835 (cls.type_name == 'IAM Policy' or cls.type_name == 'ElastiCache')):

836 try:

837 data = [{d['Name']: d['Value']} for d in data]

838 except KeyError:

839 raise PolicyValidationError(

840 f"{cls.type_name} Query Invalid Format. "

841 f"Query: {data} is not a list of key-value pairs "

842 f"from {cls.QuerySchema}"

843 )

844

845 # Support ebs-snapshot and volume 'Name', 'Values' queries without 'Filters' key

846 elif (data[0].get('Values') and

847 (cls.type_name == 'EBS Snapshot' or

848 cls.type_name == 'EBS Volume')):

849 data = [{"Filters": data}]

850

851 results = []

852 names = set()

853 for d in data:

854 if not isinstance(d, dict):

855 raise PolicyValidationError(

856 f"Query Invalid Format. Must be a list of key-value pairs "

857 f"from {cls.QuerySchema}"

858 )

859 if not len(list(d.keys())) == 1:

860 raise PolicyValidationError(

861 f"Query Invalid Format. Must be a list of key-value pairs "

862 f"from {cls.QuerySchema}"

863 )

864

865 if d.get("Filters"):

866 results.append({"Filters": cls.parse_qfilters(d["Filters"])})

867 else:

868 key, value = cls.parse_query(d)

869

870 # Allow for multiple queries with the same key

871 if key in names and (not cls.multi_value or key in cls.single_value_fields):

872 raise PolicyValidationError(

873 f"{cls.type_name} Query Invalid Key: {key} Must be unique")

874 elif key in names:

875 for q in results:

876 if list(q.keys())[0] == key:

877 q[key].append(d[key])

878 else:

879 names.add(key)

880 results.append({key: value})

881

882 return results

883

884 @classmethod

885 def parse_qfilters(cls, data):

886 if not isinstance(data, (tuple, list)):

887 raise PolicyValidationError(

888 f"{cls.type_name} Query Filter Invalid Format, must be array of dicts"

889 )

890

891 results = []

892 names = set()

893 for f in data:

894 if not isinstance(f, dict):

895 raise PolicyValidationError(

896 f"{cls.type_name} Query Filter Invalid Format, must be array of dicts"

897 )

898 if "Name" not in f or "Values" not in f:

899 raise PolicyValidationError(

900 f"{cls.type_name} Query Filter Invalid: Each filter must "

901 "contain 'Name' and 'Values' keys."

902 )

903

904 key = f['Name']

905 values = f['Values']

906

907 if key not in cls.QuerySchema.get("Filters", {}) and not key.startswith('tag:'):

908 raise PolicyValidationError(

909 f"{cls.type_name} Query Filter Invalid Key: {key} "

910 f"Valid: {', '.join(cls.QuerySchema.keys())}"

911 )

912

913 if not isinstance(values, list):

914 raise PolicyValidationError(

915 f"{cls.type_name} Query Filter Invalid Value {f} for key {key}, must be array.")

916

917 vtype = cls.QuerySchema["Filters"].get(key)

918 if vtype is None and key.startswith('tag'):

919 vtype = str

920

921 for v in values:

922 cls.type_check(vtype, v)

923

924 # Allow for multiple queries with the same key

925 if key in names:

926 for qf in results:

927 if qf['Name'] == key:

928 qf['Values'].extend(values)

929 else:

930 names.add(key)

931 results.append({'Name': key, 'Values': values})

932

933 return results

934

935 @classmethod

936 def parse_query(cls, data):

937 key = list(data.keys())[0]

938 values = list(data.values())[0]

939

940 if (not cls.multi_value or key in cls.single_value_fields) and isinstance(values, list):

941 raise PolicyValidationError(

942 f"{cls.type_name} Query Invalid Value {values}: Value for {key} must be scalar"

943 )

944 elif (cls.multi_value and key not in cls.single_value_fields

945 and not isinstance(values, list)):

946 values = [values]

947

948 if key not in cls.QuerySchema:

949 raise PolicyValidationError(

950 f"{cls.type_name} Query Invalid Key: {key} "

951 f"Valid: {', '.join(cls.QuerySchema.keys())}"

952 )

953

954 vtype = cls.QuerySchema.get(key)

955 if isinstance(values, list):

956 for v in values:

957 cls.type_check(vtype, v)

958 else:

959 cls.type_check(vtype, values)

960

961 return key, values

962

963 @classmethod

964 def type_check(cls, vtype, value):

965 if isinstance(vtype, tuple):

966 if value not in vtype:

967 raise PolicyValidationError(

968 f"{cls.type_name} Query Invalid Value: {value} Valid: {', '.join(vtype)}")

969 elif vtype == 'date':

970 if not parse_date(value):

971 raise PolicyValidationError(

972 f"{cls.type_name} Query Invalid Date Value: {value}")

973 elif not isinstance(value, vtype):

974 raise PolicyValidationError(

975 f"{cls.type_name} Query Invalid Value Type {value}"

976 )

977

978

979def get_annotation_prefix(s):

980 return 'c7n:{}'.format(s)

981

982

983def merge_dict_list(dict_iter):

984 """take an list of dictionaries and merge them.

985

986 last dict wins/overwrites on keys.

987 """

988 result = {}

989 for d in dict_iter:

990 result.update(d)

991 return result

992

993

994def merge_dict(a, b):

995 """Perform a merge of dictionaries A and B

996

997 Any subdictionaries will be recursively merged.

998 Any leaf elements in the form of scalar will use the value from B.

999 If A is a str and B is a list, A will be inserted into the front of the list.

1000 If A is a list and B is a str, B will be appended to the list.

1001 If there are two lists for the same key, the lists will be merged

1002 deduplicated with values in A first, followed by any additional values from B.

1003 """

1004 d = copy.deepcopy(a)

1005 for k, v in b.items():

1006 if k not in d:

1007 d[k] = v

1008 elif isinstance(d[k], dict) and isinstance(v, dict):

1009 d[k] = merge_dict(d[k], v)

1010 elif isinstance(d[k], list) and isinstance(v, list):

1011 for val in v:

1012 if val not in d[k]:

1013 d[k].append(val)

1014 elif isinstance(v, str) and isinstance(d[k], list):

1015 if v in d[k]:

1016 continue

1017 else:

1018 d[k].append(v)

1019 elif isinstance(v, list) and isinstance(d[k], str):

1020 if d[k] in v:

1021 d[k] = v

1022 else:

1023 d[k] = [d[k]]

1024 d[k].extend(v)

1025 elif k in d and isinstance(v, (int, str, float, bool)):

1026 d[k] = v

1027 else:

1028 raise Exception(f"k={k}, {type(v)} and {type(d[k])} not conformable.")

1029 return d

1030

1031

1032def compare_dicts_using_sets(a, b) -> bool:

1033 """Compares two dicts and replaces any lists or strings with sets

1034

1035 Compares any lists in the dict as sets.

1036 """

1037

1038 if a.keys() != b.keys():

1039 return False

1040

1041 for k, v in b.items():

1042 if isinstance(v, list):

1043 v = format_to_set(v)

1044 if isinstance(a[k], str):

1045 a[k] = format_to_set(a[k])

1046 if isinstance(a[k], list):

1047 a[k] = format_to_set(a[k])

1048 if isinstance(v, str):

1049 v = format_to_set(v)

1050 if isinstance(a[k], dict) and isinstance(v, dict):

1051 if compare_dicts_using_sets(a[k], v):

1052 continue

1053 if v != a[k]:

1054 return False

1055 return True

1056

1057

1058def format_to_set(x) -> set:

1059 """Formats lists and strings to sets.

1060

1061 Strings return as a set with one string.

1062 Lists return as a set.

1063 Variables of other datatypes will return as the original datatype.

1064 """

1065 if isinstance(x, str):

1066 return set([x])

1067 if isinstance(x, list):

1068 return set(x)

1069 else:

1070 return x

1071

1072

1073def format_dict_with_sets(x: dict) -> dict:

1074 """Formats string and list values in a dict to sets.

1075

1076 Any string value returns as a set with one string.

1077 Any list values return as a set.

1078 Returns a formatted dict.

1079 """

1080 if isinstance(x, dict):

1081 format_dict = {}

1082 for key, value in x.items():

1083 if isinstance(value, dict):

1084 format_dict[key] = format_dict_with_sets(value)

1085 else:

1086 format_dict[key] = format_to_set(value)

1087 return format_dict

1088 else:

1089 return x

1090

1091

1092def select_keys(d, keys):

1093 result = {}

1094 for k in keys:

1095 result[k] = d.get(k)

1096 return result

1097

1098

1099def get_human_size(size, precision=2):

1100 # interesting discussion on 1024 vs 1000 as base

1101 # https://en.wikipedia.org/wiki/Binary_prefix

1102 suffixes = ['B', 'KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB']

1103 suffixIndex = 0

1104 while size >= 1024 and suffixIndex < len(suffixes) - 1:

1105 suffixIndex += 1

1106 size = size / 1024.0

1107

1108 return "%.*f %s" % (precision, size, suffixes[suffixIndex])

1109

1110

1111def is_not_found(err):

1112 """Attempt for an aws exception to determine if its a NotFound error.

1113

1114 Returns boolean.

1115

1116 Across the set of AWS services the error handling behavior, runs

1117 across many different behavior patterns for NotFound style

1118 exceptions. We want to use any unambigious signal in the error

1119 code but also not flag an exception that could potentially represent

1120 another error that user should be informed about.

1121 """

1122 code = err.response['Error']['Code']

1123 for s in ('NotFound', 'NoSuch'):

1124 if s in code:

1125 return True

1126 return False

1127

1128

1129def get_support_region(manager):

1130 # support is a unique service in that it doesnt support regional endpoints

1131 # thus, we need to construct the client based off the regions found here:

1132 # https://docs.aws.amazon.com/general/latest/gr/awssupport.html

1133 #

1134 # aws-cn uses cn-north-1 for both the Beijing and Ningxia regions

1135 # https://docs.amazonaws.cn/en_us/aws/latest/userguide/endpoints-Beijing.html

1136 # https://docs.amazonaws.cn/en_us/aws/latest/userguide/endpoints-Ningxia.html

1137

1138 partition = get_partition(manager.config.region)

1139 support_region = None

1140 if partition == "aws":

1141 support_region = "us-east-1"

1142 elif partition == "aws-us-gov":

1143 support_region = "us-gov-west-1"

1144 elif partition == "aws-cn":

1145 support_region = "cn-north-1"

1146 return support_region

1147

1148

1149def get_resource_tagging_region(resource_type, region):

1150 # For global resources, tags don't populate in the get_resources call

1151 # unless the call is being made to us-east-1. For govcloud this is us-gov-west-1.

1152

1153 partition = get_partition(region)

1154 if partition == "aws":

1155 return getattr(resource_type, 'global_resource', None) and 'us-east-1' or region

1156 elif partition == "aws-us-gov":

1157 return getattr(resource_type, 'global_resource', None) and 'us-gov-west-1' or region

1158 return region

1159

1160

1161def get_eni_resource_type(eni):

1162 if eni.get('Attachment'):

1163 instance_id = eni['Attachment'].get('InstanceId')

1164 else:

1165 instance_id = None

1166 description = eni.get('Description')

1167 # EC2

1168 if instance_id:

1169 rtype = 'ec2'

1170 # ELB/ELBv2

1171 elif description.startswith('ELB app/'):

1172 rtype = 'elb-app'

1173 elif description.startswith('ELB net/'):

1174 rtype = 'elb-net'

1175 elif description.startswith('ELB gwy/'):

1176 rtype = 'elb-gwy'

1177 elif description.startswith('ELB'):

1178 rtype = 'elb'

1179 # Other Resources

1180 elif description == 'ENI managed by APIGateway':

1181 rtype = 'apigw'

1182 elif description.startswith('AWS CodeStar Connections'):

1183 rtype = 'codestar'

1184 elif description.startswith('DAX'):

1185 rtype = 'dax'

1186 elif description.startswith('AWS created network interface for directory'):

1187 rtype = 'dir'

1188 elif description == 'DMSNetworkInterface':

1189 rtype = 'dms'

1190 elif description.startswith('arn:aws:ecs:'):

1191 rtype = 'ecs'

1192 elif description.startswith('EFS mount target for'):

1193 rtype = 'fsmt'

1194 elif description.startswith('ElastiCache'):

1195 rtype = 'elasticache'

1196 elif description.startswith('AWS ElasticMapReduce'):

1197 rtype = 'emr'

1198 elif description.startswith('CloudHSM Managed Interface'):

1199 rtype = 'hsm'

1200 elif description.startswith('CloudHsm ENI'):

1201 rtype = 'hsmv2'

1202 elif description.startswith('AWS Lambda VPC ENI'):

1203 rtype = 'lambda'

1204 elif description.startswith('AWS Lambda VPC'):

1205 rtype = 'lambda'

1206 elif description.startswith('Interface for NAT Gateway'):

1207 rtype = 'nat'

1208 elif (description == 'RDSNetworkInterface' or

1209 description.startswith('Network interface for DBProxy')):

1210 rtype = 'rds'

1211 elif description == 'RedshiftNetworkInterface':

1212 rtype = 'redshift'

1213 elif description.startswith('Network Interface for Transit Gateway Attachment'):

1214 rtype = 'tgw'

1215 elif description.startswith('VPC Endpoint Interface'):

1216 rtype = 'vpce'

1217 elif description.startswith('aws-k8s-branch-eni'):

1218 rtype = 'eks'

1219 else:

1220 rtype = 'unknown'

1221 return rtype

1222

1223

1224class C7NJmespathFunctions(functions.Functions):

1225 @functions.signature(

1226 {'types': ['string']}, {'types': ['string']}

1227 )

1228 def _func_split(self, sep, string):

1229 return string.split(sep)

1230

1231 @functions.signature(

1232 {'types': ['string']}

1233 )

1234 def _func_from_json(self, string):

1235 try:

1236 return json.loads(string)

1237 except json.JSONDecodeError:

1238 return None

1239

1240

1241class C7NJMESPathParser(Parser):

1242 def parse(self, expression):

1243 result = super().parse(expression)

1244 return ParsedResultWithOptions(

1245 expression=result.expression,

1246 parsed=result.parsed

1247 )

1248

1249

1250class ParsedResultWithOptions(ParsedResult):

1251 def search(self, value, options=None):

1252 # if options are explicitly passed in, we honor those

1253 if not options:

1254 options = jmespath.Options(custom_functions=C7NJmespathFunctions())

1255 return super().search(value, options)

1256

1257

1258def jmespath_search(*args, **kwargs):

1259 return jmespath.search(

1260 *args,

1261 **kwargs,

1262 options=jmespath.Options(custom_functions=C7NJmespathFunctions())

1263 )

1264

1265

1266def get_path(path: str, resource: dict):

1267 """

1268 This function provides a wrapper to obtain a value from a resource

1269 in an efficient manner.

1270 jmespath_search is expensive and it's rarely the case that

1271 there is a path in the id field, therefore this wrapper is an optimisation.

1272

1273 :param path: the path or field name to fetch

1274 :param resource: the resource instance description

1275 :return: the field/path value

1276 """

1277 if '.' in path:

1278 return jmespath_search(path, resource)

1279 return resource[path]

1280

1281

1282def jmespath_compile(expression):

1283 parsed = C7NJMESPathParser().parse(expression)

1284 return parsed

1285

1286

1287def snap_to_period_start(start: datetime, end: datetime, period_start: str):

1288 """

1289 Adjust the start and end timestamps according to `period_start`.

1290

1291 Args:

1292 start (datetime): The original start datetime.

1293 end (datetime): The original end datetime.

1294 period_start (str): One of 'auto', 'start-of-day'.

1295

1296 Returns:

1297 Tuple[datetime, datetime]: Adjusted (start, end)

1298 """

1299 if period_start == "start-of-day":

1300 start = start.replace(hour=0, minute=0, second=0, microsecond=0)

1301 end = end.replace(hour=0, minute=0, second=0, microsecond=0)

1302 # 'auto' does nothing

1303 return start, end