Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_analysis_jobs_cypher_syntax(neo4j_session):
parameters = {
'AWS_ID': None,
'UPDATE_TAG': None,
'OKTA_ORG_ID': None,
}
for job_name in contents('cartography.data.jobs.analysis'):
if not job_name.endswith('.json'):
continue
try:
cartography.util.run_analysis_job(job_name, neo4j_session, parameters)
except Exception as e:
pytest.fail(f"run_analysis_job failed for analysis job '{job_name}' with exception: {e}")
SET a.lastupdated = {UpdateTag}
WITH a,acl MATCH (s3:S3Bucket{name: acl.bucket})
MERGE (a)-[r:APPLIES_TO]->(s3)
ON CREATE SET r.firstseen = timestamp()
SET r.lastupdated = {UpdateTag}
"""
neo4j_session.run(
ingest_acls,
acls=acls,
UpdateTag=update_tag,
)
# implement the acl permission
# https://docs.aws.amazon.com/AmazonS3/latest/dev/acl-overview.html#permissions
run_analysis_job(
'aws_s3acl_analysis.json',
neo4j_session,
{'AWS_ID': aws_account_id},
)
try:
regions = ec2.get_ec2_regions(boto3_session)
except botocore.exceptions.ClientError as e:
logger.debug("Error occurred getting EC2 regions.", exc_info=True)
logger.error(
(
"Failed to retrieve AWS region list, an error occurred: %s. The AWS sync cannot run without a valid "
"region list."
),
e,
)
return
_sync_multiple_accounts(neo4j_session, aws_accounts, regions, config.update_tag, common_job_parameters)
run_analysis_job(
'aws_ec2_asset_exposure.json',
neo4j_session,
common_job_parameters,
)
"that the identity you are authenticating to has the securityReviewer role attached."
),
e,
)
return
resources = _initialize_resources(credentials)
# If we don't have perms to pull Orgs or Folders from GCP, we will skip safely
crm.sync_gcp_organizations(neo4j_session, resources.crm_v1, config.update_tag, common_job_parameters)
crm.sync_gcp_folders(neo4j_session, resources.crm_v2, config.update_tag, common_job_parameters)
projects = crm.get_gcp_projects(resources.crm_v1)
_sync_multiple_projects(neo4j_session, resources, projects, config.update_tag, common_job_parameters)
run_analysis_job(
'gcp_compute_asset_inet_exposure.json',
neo4j_session,
common_job_parameters,
)