task: metadata: install: pip install -e .
patch: diff --git a/cartography/data/jobs/analysis/aws_foreign_accounts.json b/cartography/data/jobs/analysis/aws_foreign_accounts.json
new file mode 100644
index 00000000..fa958a6a
--- /dev/null
+++ b/cartography/data/jobs/analysis/aws_foreign_accounts.json
@@ -0,0 +1,15 @@
+{
+  "statements": [
+    {
+      "__comment": "This analyze AWS accounts we created and tag the ones that are foreign. Foreign accounts are ones that were not in the sync scope",
+      "query": "MATCH (foreign:AWSAccount) where NOT EXISTS(foreign.inscope) SET foreign.foreign = true",
+      "iterative": false
+    },
+    {
+      "__comment": "Remove accounts that were set with foreign and inscope. This can happen as we finish the list of sync accounts through assume role mapping and vpc peering",
+      "query": "MATCH (a:AWSAccount) where EXISTS(a.inscope) AND EXISTS(a.foreign) REMOVE a.foreign",
+      "iterative": false
+    }
+  ],
+  "name": "AWS - Foreign account analysis"
+}
diff --git a/cartography/data/jobs/cleanup/aws_account_cleanup.json b/cartography/data/jobs/cleanup/aws_account_cleanup.json
new file mode 100644
index 00000000..9634e4db
--- /dev/null
+++ b/cartography/data/jobs/cleanup/aws_account_cleanup.json
@@ -0,0 +1,15 @@
+{
+  "statements": [
+    {
+      "query": "MATCH (n:AWSAccount{foreign: true}) WHERE n.inscope = true AND n.lastupdated <> {UPDATE_TAG} WITH n LIMIT {LIMIT_SIZE} REMOVE n.foreign",
+      "iterative": true,
+      "iterationsize": 100
+    },
+    {
+      "query": "MATCH (n:AWSAccount{foreign: true}) WHERE n.lastupdated <> {UPDATE_TAG} WITH n LIMIT {LIMIT_SIZE} DETACH DELETE (n)",
+      "iterative": true,
+      "iterationsize": 100
+    }
+  ],
+  "name": "AWS accounts cleanup"
+}
diff --git a/cartography/data/jobs/cleanup/aws_import_principals_cleanup.json b/cartography/data/jobs/cleanup/aws_import_principals_cleanup.json
index b98736b9..2f9ebf01 100644
--- a/cartography/data/jobs/cleanup/aws_import_principals_cleanup.json
+++ b/cartography/data/jobs/cleanup/aws_import_principals_cleanup.json
@@ -21,11 +21,10 @@
       "iterationsize": 100
     },
     {
-    "query": "MATCH (:AWSAccount{id: $AWS_ID})-[:RESOURCE]->(n:AWSPrincipal) WHERE n.lastupdated <> $UPDATE_TAG WITH n LIMIT $LIMIT_SIZE DETACH DELETE (n)",
+    "query": "MATCH (AWSAccount{foreign:true})-[:RESOURCE]->(n:AWSPrincipal) WHERE n.lastupdated <> {UPDATE_TAG} WITH n LIMIT {LIMIT_SIZE} DETACH DELETE (n)",
     "iterative": true,
     "iterationsize": 100
-  }
-
+    }
 ],
   "name": "cleanup AWSPrincipals, AWSPolicies and AWSPolicyStatements"
 }
diff --git a/cartography/intel/aws/__init__.py b/cartography/intel/aws/__init__.py
index d0762994..99a8616f 100644
--- a/cartography/intel/aws/__init__.py
+++ b/cartography/intel/aws/__init__.py
@@ -261,3 +261,9 @@ def start_aws_ingestion(neo4j_session: neo4j.Session, config: Config) -> None:
             neo4j_session,
             common_job_parameters,
         )
+
+        run_analysis_job(
+            'aws_foreign_accounts.json',
+            neo4j_session,
+            common_job_parameters,
+        )
diff --git a/cartography/intel/aws/ec2/network_interfaces.py b/cartography/intel/aws/ec2/network_interfaces.py
index e4ca2da4..bb68a615 100644
--- a/cartography/intel/aws/ec2/network_interfaces.py
+++ b/cartography/intel/aws/ec2/network_interfaces.py
@@ -82,9 +82,8 @@ def load_network_interfaces(
             ON CREATE SET r.firstseen = timestamp()
             SET r.lastupdated = $update_tag
         WITH network_interface, netinf
-
         MERGE (acc:AWSAccount{id: $aws_account_id})
-        ON CREATE SET acc.firstseen = timestamp()
+        ON CREATE SET acc.firstseen = timestamp(), acc.inscope=true
         SET acc.lastupdated = $update_tag
         MERGE (acc)-[r:RESOURCE]->(netinf)
         ON CREATE SET r.firstseen = timestamp()
diff --git a/cartography/intel/aws/ec2/tgw.py b/cartography/intel/aws/ec2/tgw.py
index c15ea347..340d8797 100644
--- a/cartography/intel/aws/ec2/tgw.py
+++ b/cartography/intel/aws/ec2/tgw.py
@@ -74,7 +74,7 @@ def load_transit_gateways(
 ) -> None:
     ingest_transit_gateway = """
     MERGE (ownerAccount:AWSAccount {id: $OwnerId})
-    ON CREATE SET ownerAccount.firstseen = timestamp(), ownerAccount.foreign = true
+    ON CREATE SET ownerAccount.firstseen = timestamp()
     SET ownerAccount.lastupdated = $update_tag
 
     MERGE (tgw:AWSTransitGateway {id: $ARN})
diff --git a/cartography/intel/aws/ec2/vpc_peerings.py b/cartography/intel/aws/ec2/vpc_peerings.py
index 42ed8ed0..93e11a57 100644
--- a/cartography/intel/aws/ec2/vpc_peerings.py
+++ b/cartography/intel/aws/ec2/vpc_peerings.py
@@ -51,11 +51,11 @@ def load_vpc_peerings(
     SET rvpc.lastupdated = $update_tag, rvpc.vpcid = vpc_peering.RequesterVpcInfo.VpcId
 
     MERGE (aaccount:AWSAccount{id: vpc_peering.AccepterVpcInfo.OwnerId})
-    ON CREATE SET aaccount.firstseen = timestamp(), aaccount.foreign = true
+    ON CREATE SET aaccount.firstseen = timestamp()
     SET aaccount.lastupdated = $update_tag
 
     MERGE (raccount:AWSAccount{id: vpc_peering.RequesterVpcInfo.OwnerId})
-    ON CREATE SET raccount.firstseen = timestamp(), raccount.foreign = true
+    ON CREATE SET raccount.firstseen = timestamp()
     SET raccount.lastupdated = $update_tag
 
     MERGE (pcx)-[rav:ACCEPTER_VPC]->(avpc)
diff --git a/cartography/intel/aws/iam.py b/cartography/intel/aws/iam.py
index 3f95639e..90662b94 100644
--- a/cartography/intel/aws/iam.py
+++ b/cartography/intel/aws/iam.py
@@ -15,6 +15,7 @@ from cartography.stats import get_stats_client
 from cartography.util import merge_module_sync_metadata
 from cartography.util import run_cleanup_job
 from cartography.util import timeit
+
 logger = logging.getLogger(__name__)
 stat_handler = get_stats_client(__name__)
 
@@ -305,6 +306,25 @@ def load_roles(
     SET r.lastupdated = $aws_update_tag
     """
 
+    # Note - why we don't set inscope or foreign attribute on the account
+    #
+    # we are agnostic here if this is the AWSAccount is part of the sync scope or
+    # a foreign AWS account that contains a trusted principal. The account could also be inscope
+    # but not sync yet.
+    # - The inscope attribute - set when the account is being sync.
+    # - The foreign attribute - the attribute assignment logic is in aws_foreign_accounts.json analysis job
+    # - Why seperate statement is needed - the arn may point to service level principals ex - ec2.amazonaws.com
+    ingest_spnmap_statement = """
+    MERGE (aa:AWSAccount{id: $SpnAccountId})
+    ON CREATE SET aa.firstseen = timestamp()
+    SET aa.lastupdated = $aws_update_tag
+    WITH aa
+    MATCH (spnnode:AWSPrincipal{arn: $SpnArn})
+    WITH spnnode, aa
+    MERGE (aa)-[r:RESOURCE]->(spnnode)
+    ON CREATE SET r.firstseen = timestamp()
+    """
+
     # TODO support conditions
     logger.info(f"Loading {len(roles)} IAM roles to the graph.")
     for role in roles:
@@ -329,6 +349,14 @@ def load_roles(
                     RoleArn=role['Arn'],
                     aws_update_tag=aws_update_tag,
                 )
+                spn_arn = get_account_from_arn(principal_value)
+                if spn_arn:
+                    neo4j_session.run(
+                        ingest_spnmap_statement,
+                        SpnArn=principal_value,
+                        SpnAccountId=get_account_from_arn(principal_value),
+                        aws_update_tag=aws_update_tag,
+                    )
 
 
 @timeit
@@ -739,3 +767,19 @@ def sync(
         update_tag=update_tag,
         stat_handler=stat_handler,
     )
+
+
+@timeit
+def get_account_from_arn(arn: str) -> str:
+    # ARN documentation
+    # https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html
+
+    if not arn.startswith("arn:"):
+        # must be a service principal arn, such as ec2.amazonaws.com
+        return ""
+
+    parts = arn.split(":")
+    if len(parts) < 4:
+        return ""
+    else:
+        return parts[4]
diff --git a/cartography/intel/aws/organizations.py b/cartography/intel/aws/organizations.py
index 270a3506..bfcbc9cb 100644
--- a/cartography/intel/aws/organizations.py
+++ b/cartography/intel/aws/organizations.py
@@ -93,7 +93,8 @@ def load_aws_accounts(
     query = """
     MERGE (aa:AWSAccount{id: $ACCOUNT_ID})
     ON CREATE SET aa.firstseen = timestamp()
-    SET aa.lastupdated = $aws_update_tag, aa.name = $ACCOUNT_NAME
+    SET aa.lastupdated = $aws_update_tag, aa.name = $ACCOUNT_NAME, aa.inscope=true
+    REMOVE aa.foreign
     WITH aa
     MERGE (root:AWSPrincipal{arn: $RootArn})
     ON CREATE SET root.firstseen = timestamp(), root.type = 'AWS'
diff --git a/docs/root/modules/aws/schema.md b/docs/root/modules/aws/schema.md
index f890a8e8..8ae15835 100644
--- a/docs/root/modules/aws/schema.md
+++ b/docs/root/modules/aws/schema.md
@@ -10,6 +10,8 @@ Representation of an AWS Account.
 |-------|-------------|
 |firstseen| Timestamp of when a sync job discovered this node|
 |name| The name of the account|
+|inscope| Indicates that the account is part of the sync scope (true or false).
+|foreign| Indicates if the account is not part of the sync scope (true or false). One such example is an account that is trusted as part of cross-account AWSRole trust not in scope for sync.
 |lastupdated| Timestamp of the last time the node was updated|
 |**id**| The AWS Account ID number|
 

problem_statement: Missing AWS - RESOURCE -> AWSPrincipal 
*Description*:
> What issue is being seen?  When ingesting policy, we create AWSPrincipal but we don't create and associate it to it's respective AWSAccount. The AWSPrincipal can be in the account being sync but can also be part of another account that is trusted.

https://github.com/lyft/cartography/blob/8db4a7eb1b57df8e681eeef09523a34e963aa09b/cartography/intel/aws/iam.py#L297
    ingest_policy_statement = """
    MERGE (spnnode:AWSPrincipal{arn: {SpnArn}})
    ON CREATE SET spnnode.firstseen = timestamp()
    SET spnnode.lastupdated = {aws_update_tag}, spnnode.type = {SpnType}
    WITH spnnode
    MATCH (role:AWSRole{arn: {RoleArn}})
    MERGE (role)-[r:TRUSTS_AWS_PRINCIPAL]->(spnnode)
    ON CREATE SET r.firstseen = timestamp()
    SET r.lastupdated = {aws_update_tag}
    """

*To Reproduce*:
Create a trust relationship to another AWS account that is not part of the scanning. Even if the AWSPrincipal is part of the scanned account, the RESOURCE won't be created (see code)

*Logs*:
N/A - no error is generated

*Screenshots*:
> If applicable, add screenshots to help explain your problem.

*Please complete the following information:*:
>  - Cartography release version : 0.65.0
>  - Python version: [e.g. 3.7.4] 3.10

repo: lyft/cartography
test_cmd: pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
test_patch: diff --git a/tests/integration/cartography/intel/aws/test_init.py b/tests/integration/cartography/intel/aws/test_init.py
index b483beb0..d94edcbb 100644
--- a/tests/integration/cartography/intel/aws/test_init.py
+++ b/tests/integration/cartography/intel/aws/test_init.py
@@ -92,7 +92,7 @@ def test_start_aws_ingestion(mock_run_analysis, mock_sync_multiple, mock_orgs, m
     assert mock_sync_multiple.call_count == 1
 
     # Brittle, but here to ensure that our mock_run_analysis path is correct.
-    assert mock_run_analysis.call_count == 3
+    assert mock_run_analysis.call_count == 4
 
 
 @mock.patch('cartography.intel.aws.boto3.Session')
@@ -162,7 +162,7 @@ def test_start_aws_ingestion_does_cleanup(
     mock_get_aws_account.return_value = {'test_profile': 'test_account', 'test_profile2': 'test_account2'}
     cartography.intel.aws.start_aws_ingestion(neo4j_session, test_config)
 
-    assert mock_run_analysis.call_count == 3
+    assert mock_run_analysis.call_count == 4
     assert mock_run_cleanup_job.call_count == 1
 
 
diff --git a/tests/unit/cartography/intel/aws/test_iam.py b/tests/unit/cartography/intel/aws/test_iam.py
index 30beb9a2..47353afb 100644
--- a/tests/unit/cartography/intel/aws/test_iam.py
+++ b/tests/unit/cartography/intel/aws/test_iam.py
@@ -30,3 +30,8 @@ def test__parse_principal_entries():
     assert principal_entries[1] == ("Service", "test-service-1")
     assert principal_entries[2] == ("Service", "test-service-2")
     assert principal_entries[3] == ("Federated", "test-provider-1")
+
+
+def test_get_account_from_arn():
+    result = iam.get_account_from_arn("arn:aws:iam::081157660428:role/TestRole")
+    assert result == "081157660428"