sfn-secrets-rotation-chain

============================= test session starts ==============================
platform linux -- Python 3.12.3, pytest-8.4.1, pluggy-1.6.0 -- /root/.cache/uv/archive-v0/Eklpkw38n76eq7Qn9jL4r/bin/python
cachedir: .pytest_cache
rootdir: /tests
plugins: json-ctrf-0.3.5
collecting ... collected 15 items

../tests/test_state.py::test_aws_cli_binary_is_real PASSED               [  6%]
../tests/test_state.py::test_localstack_reachable PASSED                 [ 13%]
../tests/test_state.py::test_secret_exists_and_uses_customer_cmk PASSED  [ 20%]
../tests/test_state.py::test_rotation_lambda_exists PASSED               [ 26%]
../tests/test_state.py::test_notification_lambda_exists PASSED           [ 33%]
../tests/test_state.py::test_parent_and_child_state_machines_exist PASSED [ 40%]
../tests/test_state.py::test_event_bus_and_rule_exist PASSED             [ 46%]
../tests/test_state.py::test_rule_target_points_at_notification_lambda PASSED [ 53%]
../tests/test_state.py::test_rotator_role_grants_kms_actions PASSED      [ 60%]
../tests/test_state.py::test_kms_key_policy_grants_rotator_role FAILED   [ 66%]
../tests/test_state.py::test_rotator_lambda_allows_secretsmanager_invoke FAILED [ 73%]
../tests/test_state.py::test_parent_role_grants_sync2_managed_rule_perms FAILED [ 80%]
../tests/test_state.py::test_child_role_can_rotate_and_publish FAILED    [ 86%]
../tests/test_state.py::test_notification_lambda_allows_eventbridge_invoke PASSED [ 93%]
../tests/test_state.py::test_end_to_end_parent_execution_rotates_and_notifies PASSED [100%]

=================================== FAILURES ===================================
___________________ test_kms_key_policy_grants_rotator_role ____________________

iam = <botocore.client.IAM object at 0xffff7d4cdf40>
kms = <botocore.client.KMS object at 0xffff7da18320>

    def test_kms_key_policy_grants_rotator_role(iam, kms):
        role_arn = iam.get_role(RoleName=ROTATOR_ROLE)["Role"]["Arn"]
        policy_str = kms.get_key_policy(KeyId=KEY_ALIAS, PolicyName="default")[
            "Policy"
        ]
        policy = json.loads(policy_str)
        match = False
        for st in policy.get("Statement", []):
            if st.get("Effect") != "Allow":
                continue
            principal = st.get("Principal") or {}
            if not isinstance(principal, dict):
                continue
            arns = set(_normalise_list(principal.get("AWS")))
            if role_arn not in arns:
                continue
            actions = set(_normalise_list(st.get("Action")))
            if _actions_cover(actions, REQUIRED_KMS_ACTIONS):
                match = True
                break
>       assert match, (
            f"KMS key policy on {KEY_ALIAS} has no Allow statement whose "
            f"Principal.AWS includes {role_arn} and whose Action covers "
            f"{sorted(REQUIRED_KMS_ACTIONS)}. Key policy: {policy_str}"
        )
E       AssertionError: KMS key policy on alias/app-rotation-key has no Allow statement whose Principal.AWS includes arn:aws:iam::000000000000:role/secrets-rotator-role and whose Action covers ['kms:Decrypt', 'kms:GenerateDataKey']. Key policy: {"Version": "2012-10-17", "Id": "key-default-1", "Statement": [{"Sid": "Enable IAM User Permissions", "Effect": "Allow", "Principal": {"AWS": "arn:aws:iam::000000000000:root"}, "Action": "kms:*", "Resource": "*"}]}
E       assert False

/tests/test_state.py:322: AssertionError
_______________ test_rotator_lambda_allows_secretsmanager_invoke _______________

lmb = <botocore.client.Lambda object at 0xffff7da1a090>

    def test_rotator_lambda_allows_secretsmanager_invoke(lmb):
        try:
            policy_doc = json.loads(
>               lmb.get_policy(FunctionName=ROTATOR_FUNCTION)["Policy"]
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
            )

/tests/test_state.py:332: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/root/.cache/uv/archive-v0/Eklpkw38n76eq7Qn9jL4r/lib/python3.12/site-packages/botocore/client.py:569: in _api_call
    return self._make_api_call(operation_name, kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <botocore.client.Lambda object at 0xffff7da1a090>
operation_name = 'GetPolicy', api_params = {'FunctionName': 'secrets-rotator'}

    def _make_api_call(self, operation_name, api_params):
        operation_model = self._service_model.operation_model(operation_name)
        service_name = self._service_model.service_name
        history_recorder.record(
            'API_CALL',
            {
                'service': service_name,
                'operation': operation_name,
                'params': api_params,
            },
        )
        if operation_model.deprecated:
            logger.debug(
                'Warning: %s.%s() is deprecated', service_name, operation_name
            )
        request_context = {
            'client_region': self.meta.region_name,
            'client_config': self.meta.config,
            'has_streaming_input': operation_model.has_streaming_input,
            'auth_type': operation_model.resolved_auth_type,
            'unsigned_payload': operation_model.unsigned_payload,
        }
    
        api_params = self._emit_api_params(
            api_params=api_params,
            operation_model=operation_model,
            context=request_context,
        )
        (
            endpoint_url,
            additional_headers,
            properties,
        ) = self._resolve_endpoint_ruleset(
            operation_model, api_params, request_context
        )
        if properties:
            # Pass arbitrary endpoint info with the Request
            # for use during construction.
            request_context['endpoint_properties'] = properties
        request_dict = self._convert_to_request_dict(
            api_params=api_params,
            operation_model=operation_model,
            endpoint_url=endpoint_url,
            context=request_context,
            headers=additional_headers,
        )
        resolve_checksum_context(request_dict, operation_model, api_params)
    
        service_id = self._service_model.service_id.hyphenize()
        handler, event_response = self.meta.events.emit_until_response(
            f'before-call.{service_id}.{operation_name}',
            model=operation_model,
            params=request_dict,
            request_signer=self._request_signer,
            context=request_context,
        )
    
        if event_response is not None:
            http, parsed_response = event_response
        else:
            maybe_compress_request(
                self.meta.config, request_dict, operation_model
            )
            apply_request_checksum(request_dict)
            http, parsed_response = self._make_request(
                operation_model, request_dict, request_context
            )
    
        self.meta.events.emit(
            f'after-call.{service_id}.{operation_name}',
            http_response=http,
            parsed=parsed_response,
            model=operation_model,
            context=request_context,
        )
    
        if http.status_code >= 300:
            error_info = parsed_response.get("Error", {})
            error_code = error_info.get("QueryErrorCode") or error_info.get(
                "Code"
            )
            error_class = self.exceptions.from_code(error_code)
>           raise error_class(parsed_response, operation_name)
E           botocore.errorfactory.ResourceNotFoundException: An error occurred (ResourceNotFoundException) when calling the GetPolicy operation: The resource you requested does not exist.

/root/.cache/uv/archive-v0/Eklpkw38n76eq7Qn9jL4r/lib/python3.12/site-packages/botocore/client.py:1023: ResourceNotFoundException

During handling of the above exception, another exception occurred:

lmb = <botocore.client.Lambda object at 0xffff7da1a090>

    def test_rotator_lambda_allows_secretsmanager_invoke(lmb):
        try:
            policy_doc = json.loads(
                lmb.get_policy(FunctionName=ROTATOR_FUNCTION)["Policy"]
            )
        except ClientError as e:
>           pytest.fail(
                f"Lambda {ROTATOR_FUNCTION} has no resource-based policy , "
                f"secretsmanager.amazonaws.com cannot invoke it: {e}"
            )
E           Failed: Lambda secrets-rotator has no resource-based policy , secretsmanager.amazonaws.com cannot invoke it: An error occurred (ResourceNotFoundException) when calling the GetPolicy operation: The resource you requested does not exist.

/tests/test_state.py:335: Failed
_______________ test_parent_role_grants_sync2_managed_rule_perms _______________

iam = <botocore.client.IAM object at 0xffff7d4cdf40>

    def test_parent_role_grants_sync2_managed_rule_perms(iam):
        """The .sync:2 integration registers an EventBridge-managed callback
        rule. Parent role needs events:PutRule/PutTargets/DescribeRule plus
        full states:* on the child."""
        found_events = False
        found_states = False
        for _pname, st in _inline_statements(iam, PARENT_ROLE):
            actions = set(_normalise_list(st.get("Action")))
            if _actions_cover(actions, REQUIRED_PARENT_EVENTS_ACTIONS):
                found_events = True
            if _actions_cover(actions, REQUIRED_PARENT_STATES_ACTIONS):
                found_states = True
        assert found_events, (
            f"{PARENT_ROLE} missing events:PutRule/PutTargets/DescribeRule. "
            f".sync:2 uses an EventBridge-managed callback rule; without "
            f"these the parent cannot set up the callback."
        )
>       assert found_states, (
            f"{PARENT_ROLE} missing states:StartExecution/DescribeExecution/"
            f"StopExecution. .sync:2 requires all three on the child SM."
        )
E       AssertionError: rotation-orchestrator-role missing states:StartExecution/DescribeExecution/StopExecution. .sync:2 requires all three on the child SM.
E       assert False

/tests/test_state.py:377: AssertionError
____________________ test_child_role_can_rotate_and_publish ____________________

iam = <botocore.client.IAM object at 0xffff7d4cdf40>

    def test_child_role_can_rotate_and_publish(iam):
        found_rotate = False
        found_putevents = False
        for _pname, st in _inline_statements(iam, CHILD_ROLE):
            actions = set(_normalise_list(st.get("Action")))
            if _actions_cover(actions, {"secretsmanager:RotateSecret"}):
                found_rotate = True
            if _actions_cover(actions, {"events:PutEvents"}):
                found_putevents = True
>       assert found_rotate, (
            f"{CHILD_ROLE} lacks secretsmanager:RotateSecret , the child "
            f"state machine cannot trigger rotation."
        )
E       AssertionError: rotation-child-role lacks secretsmanager:RotateSecret , the child state machine cannot trigger rotation.
E       assert False

/tests/test_state.py:392: AssertionError
=============================== warnings summary ===============================
test_state.py: 33 warnings
  /root/.cache/uv/archive-v0/Eklpkw38n76eq7Qn9jL4r/lib/python3.12/site-packages/botocore/auth.py:424: DeprecationWarning: datetime.datetime.utcnow() is deprecated and scheduled for removal in a future version. Use timezone-aware objects to represent datetimes in UTC: datetime.datetime.now(datetime.UTC).
    datetime_now = datetime.datetime.utcnow()

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
==================================== PASSES ====================================
=========================== short test summary info ============================
PASSED ../tests/test_state.py::test_aws_cli_binary_is_real
PASSED ../tests/test_state.py::test_localstack_reachable
PASSED ../tests/test_state.py::test_secret_exists_and_uses_customer_cmk
PASSED ../tests/test_

… (truncated at 12,000 chars, full verifier log is in the trial artifacts)