EC2: Cross-account transit gateway peering attachments (#6851)

moto/ec2/exceptions.py

@@ -436,6 +436,14 @@ def __init__(self, parameter_value: str):
         )
 
 
+class InvalidParameterValueErrorPeeringAttachment(EC2ClientError):
+    def __init__(self, operation: str, transit_gateway_attachment_id: str):
+        super().__init__(
+            "InvalidParameterValue",
+            f"Cannot {operation} {transit_gateway_attachment_id} as the source of the peering request.",
+        )
+
+
 class InvalidParameterValueErrorTagSpotFleetRequest(EC2ClientError):
     def __init__(self, resource_type: str):
         super().__init__(

moto/ec2/models/transit_gateway_attachments.py

@@ -1,8 +1,11 @@
-from typing import Any, Dict, List, Optional
+from collections import defaultdict
+import weakref
+from typing import Any, Dict, List, Optional, Iterator
 from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow
 from moto.utilities.utils import merge_multiple_dicts, filter_resources
 from .core import TaggedEC2Resource
 from .vpc_peering_connections import PeeringConnectionStatus
+from ..exceptions import InvalidParameterValueErrorPeeringAttachment
 from ..utils import random_transit_gateway_attachment_id, describe_tag_filter
 
 
@@ -99,8 +102,20 @@ def __init__(
 
 
 class TransitGatewayAttachmentBackend:
+    backend_refs = defaultdict(set)  # type: ignore
+
     def __init__(self) -> None:
         self.transit_gateway_attachments: Dict[str, TransitGatewayAttachment] = {}
+        self.backend_refs[self.__class__].add(weakref.ref(self))
+
+    @classmethod
+    def _get_peering_attachment_backend_refs(
+        cls,
+    ) -> Iterator["TransitGatewayAttachmentBackend"]:
+        for backend_ref in cls.backend_refs[cls]:
+            backend = backend_ref()
+            if backend is not None:
+                yield backend
 
     def create_transit_gateway_vpn_attachment(
         self,
@@ -276,11 +291,24 @@ def create_transit_gateway_peering_attachment(
             tags=tags,
             region_name=self.region_name,  # type: ignore[attr-defined]
         )
-        transit_gateway_peering_attachment.status.accept()
-        transit_gateway_peering_attachment.state = "available"
+
         self.transit_gateway_attachments[
             transit_gateway_peering_attachment.id
         ] = transit_gateway_peering_attachment
+
+        # If the peer is not same as the current account or region, create attachment in peer backend
+        if self.account_id != peer_account_id or self.region_name != peer_region:  # type: ignore[attr-defined]
+            for backend in self._get_peering_attachment_backend_refs():
+                if (
+                    backend.account_id == peer_account_id  # type: ignore[attr-defined]
+                    and backend.region_name == peer_region  # type: ignore[attr-defined]
+                ):
+                    backend.transit_gateway_attachments[
+                        transit_gateway_peering_attachment.id
+                    ] = transit_gateway_peering_attachment
+
+        transit_gateway_peering_attachment.status.pending()
+        transit_gateway_peering_attachment.state = "pendingAcceptance"
         return transit_gateway_peering_attachment
 
     def describe_transit_gateway_peering_attachments(
@@ -319,6 +347,23 @@ def accept_transit_gateway_peering_attachment(
         transit_gateway_attachment = self.transit_gateway_attachments[
             transit_gateway_attachment_id
         ]
+
+        requester_account_id = transit_gateway_attachment.requester_tgw_info["ownerId"]  # type: ignore[attr-defined]
+        requester_region_name = transit_gateway_attachment.requester_tgw_info["region"]  # type: ignore[attr-defined]
+        accepter_account_id = transit_gateway_attachment.accepter_tgw_info["ownerId"]  # type: ignore[attr-defined]
+        accepter_region_name = transit_gateway_attachment.accepter_tgw_info["region"]  # type: ignore[attr-defined]
+
+        # For cross-account peering, must be accepted by the accepter
+        if requester_account_id != accepter_account_id and self.account_id != accepter_account_id:  # type: ignore[attr-defined]
+            raise InvalidParameterValueErrorPeeringAttachment(
+                "accept", transit_gateway_attachment_id
+            )
+
+        if requester_region_name != accepter_region_name and self.region_name != accepter_region_name:  # type: ignore[attr-defined]
+            raise InvalidParameterValueErrorPeeringAttachment(
+                "accept", transit_gateway_attachment_id
+            )
+
         transit_gateway_attachment.state = "available"
         # Bit dodgy - we just assume that we act on a TransitGatewayPeeringAttachment
         # We could just as easily have another sub-class of TransitGatewayAttachment on our hands, which does not have a status-attribute
@@ -331,6 +376,22 @@ def reject_transit_gateway_peering_attachment(
         transit_gateway_attachment = self.transit_gateway_attachments[
             transit_gateway_attachment_id
         ]
+
+        requester_account_id = transit_gateway_attachment.requester_tgw_info["ownerId"]  # type: ignore[attr-defined]
+        requester_region_name = transit_gateway_attachment.requester_tgw_info["region"]  # type: ignore[attr-defined]
+        accepter_account_id = transit_gateway_attachment.accepter_tgw_info["ownerId"]  # type: ignore[attr-defined]
+        accepter_region_name = transit_gateway_attachment.requester_tgw_info["region"]  # type: ignore[attr-defined]
+
+        if requester_account_id != accepter_account_id and self.account_id != accepter_account_id:  # type: ignore[attr-defined]
+            raise InvalidParameterValueErrorPeeringAttachment(
+                "reject", transit_gateway_attachment_id
+            )
+
+        if requester_region_name != accepter_region_name and self.region_name != accepter_region_name:  # type: ignore[attr-defined]
+            raise InvalidParameterValueErrorPeeringAttachment(
+                "reject", transit_gateway_attachment_id
+            )
+
         transit_gateway_attachment.state = "rejected"
         transit_gateway_attachment.status.reject()  # type: ignore[attr-defined]
         return transit_gateway_attachment

tests/test_ec2/test_transit_gateway_peering_attachments.py

@@ -1,7 +1,12 @@
+import os
+
 import boto3
+import pytest
+from botocore.exceptions import ClientError
+
 from moto import mock_ec2, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
-from unittest import SkipTest
+from unittest import SkipTest, mock
 
 
 @mock_ec2
@@ -89,7 +94,7 @@ def test_describe_transit_gateway_peering_attachment_by_filters():
     )["TransitGatewayPeeringAttachments"]
     assert [a["TransitGatewayAttachmentId"] for a in find_3] == [attchmnt3]
 
-    filters = [{"Name": "state", "Values": ["available"]}]
+    filters = [{"Name": "state", "Values": ["pendingAcceptance"]}]
     find_all = retrieve_all_attachments(ec2, filters)
     all_ids = [a["TransitGatewayAttachmentId"] for a in find_all]
     assert attchmnt1 in all_ids
@@ -105,7 +110,7 @@ def test_describe_transit_gateway_peering_attachment_by_filters():
 
     find_available = ec2.describe_transit_gateway_peering_attachments(
         TransitGatewayAttachmentIds=[attchmnt1, attchmnt2],
-        Filters=[{"Name": "state", "Values": ["available"]}],
+        Filters=[{"Name": "state", "Values": ["pendingAcceptance"]}],
     )["TransitGatewayPeeringAttachments"]
     assert [a["TransitGatewayAttachmentId"] for a in find_available] == [attchmnt1]
 
@@ -119,7 +124,9 @@ def test_create_and_accept_transit_gateway_peering_attachment():
     gateway_id2 = ec2.create_transit_gateway(Description="my second gateway")[
         "TransitGateway"
     ]["TransitGatewayId"]
-    attchment_id = create_peering_attachment(ec2, gateway_id1, gateway_id2)
+    attchment_id = create_peering_attachment(
+        ec2, gateway_id1, gateway_id2, peer_region="us-west-1"
+    )
 
     ec2.accept_transit_gateway_peering_attachment(
         TransitGatewayAttachmentId=attchment_id
@@ -176,12 +183,86 @@ def test_create_and_delete_transit_gateway_peering_attachment():
     assert attachment["State"] == "deleted"
 
 
-def create_peering_attachment(ec2, gateway_id1, gateway_id2):
+@mock_ec2
+@pytest.mark.parametrize(
+    "account1,account2",
+    [
+        pytest.param("111111111111", "111111111111", id="within account"),
+        pytest.param("111111111111", "222222222222", id="across accounts"),
+    ],
+)
+@pytest.mark.skipif(
+    settings.TEST_SERVER_MODE, reason="Cannot set account ID in server mode"
+)
+def test_transit_gateway_peering_attachments_cross_region(account1, account2):
+    # create transit gateways
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+        ec2_us = boto3.client("ec2", "us-west-1")
+        gateway_us = ec2_us.create_transit_gateway()["TransitGateway"][
+            "TransitGatewayId"
+        ]
+
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+        ec2_eu = boto3.client("ec2", "eu-central-1")
+        gateway_eu = ec2_eu.create_transit_gateway()["TransitGateway"][
+            "TransitGatewayId"
+        ]
+
+    # create peering
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+        attachment_id = create_peering_attachment(
+            ec2_us,
+            gateway_us,
+            gateway_eu,
+            peer_account=account2,
+            peer_region="eu-central-1",
+        )
+
+    # ensure peering can be described by the accepter
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+        response = ec2_eu.describe_transit_gateway_peering_attachments(
+            TransitGatewayAttachmentIds=[attachment_id]
+        )["TransitGatewayPeeringAttachments"][0]
+        assert response["TransitGatewayAttachmentId"] == attachment_id
+        assert response["RequesterTgwInfo"]["OwnerId"] == account1
+        assert response["RequesterTgwInfo"]["Region"] == "us-west-1"
+        assert response["AccepterTgwInfo"]["OwnerId"] == account2
+        assert response["AccepterTgwInfo"]["Region"] == "eu-central-1"
+
+    # ensure accepting in requester account/region raises
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account1}):
+        with pytest.raises(ClientError) as exc:
+            ec2_us.accept_transit_gateway_peering_attachment(
+                TransitGatewayAttachmentId=attachment_id
+            )
+        assert exc.value.response["Error"]["Code"] == "InvalidParameterValue"
+        assert (
+            exc.value.response["Error"]["Message"]
+            == f"Cannot accept {attachment_id} as the source of the peering request."
+        )
+
+    with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}):
+        # ensure peering can be accepted by the accepter
+        response = ec2_eu.accept_transit_gateway_peering_attachment(
+            TransitGatewayAttachmentId=attachment_id
+        )
+        assert response["TransitGatewayPeeringAttachment"]["State"] == "available"
+
+        # ensure peering can be deleted by the accepter
+        response = ec2_eu.delete_transit_gateway_peering_attachment(
+            TransitGatewayAttachmentId=attachment_id
+        )
+        assert response["TransitGatewayPeeringAttachment"]["State"] == "deleted"
+
+
+def create_peering_attachment(
+    ec2, gateway_id1, gateway_id2, peer_account=ACCOUNT_ID, peer_region="us-east-1"
+):
     return ec2.create_transit_gateway_peering_attachment(
         TransitGatewayId=gateway_id1,
         PeerTransitGatewayId=gateway_id2,
-        PeerAccountId=ACCOUNT_ID,
-        PeerRegion="us-east-1",
+        PeerAccountId=peer_account,
+        PeerRegion=peer_region,
     )["TransitGatewayPeeringAttachment"]["TransitGatewayAttachmentId"]