Partial port of S3ArnParamHandler from botocore. This is used to make a determination on MRAP accesspoints for signing purposes. This should be safe to remove once we properly integrate auth resolution from Botocore into the CRT transfer integration.
| 929 | |
| 930 | |
| 931 | class _S3ArnParamHandler: |
| 932 | """Partial port of S3ArnParamHandler from botocore. |
| 933 | |
| 934 | This is used to make a determination on MRAP accesspoints for signing |
| 935 | purposes. This should be safe to remove once we properly integrate auth |
| 936 | resolution from Botocore into the CRT transfer integration. |
| 937 | """ |
| 938 | |
| 939 | _RESOURCE_REGEX = re.compile( |
| 940 | r'^(?P<resource_type>accesspoint|outpost)[/:](?P<resource_name>.+)$' |
| 941 | ) |
| 942 | |
| 943 | def __init__(self): |
| 944 | self._arn_parser = ArnParser() |
| 945 | |
| 946 | def handle_arn(self, bucket): |
| 947 | arn_details = self._get_arn_details_from_bucket(bucket) |
| 948 | if arn_details is None: |
| 949 | return |
| 950 | if arn_details['resource_type'] == 'accesspoint': |
| 951 | return arn_details |
| 952 | |
| 953 | def _get_arn_details_from_bucket(self, bucket): |
| 954 | try: |
| 955 | arn_details = self._arn_parser.parse_arn(bucket) |
| 956 | self._add_resource_type_and_name(arn_details) |
| 957 | return arn_details |
| 958 | except InvalidArnException: |
| 959 | pass |
| 960 | return None |
| 961 | |
| 962 | def _add_resource_type_and_name(self, arn_details): |
| 963 | match = self._RESOURCE_REGEX.match(arn_details['resource']) |
| 964 | if match: |
| 965 | arn_details['resource_type'] = match.group('resource_type') |
| 966 | arn_details['resource_name'] = match.group('resource_name') |
no outgoing calls
no test coverage detected