v1.9.6 PROC'd Runners

.gitignore

@@ -3,6 +3,7 @@
 build/
 node_modules/
 .idea/
+.terraform/
 
 docs/doctrees
 src/webui/frontend/src/**/*.css

migrations/v1_9_5-v1_9_6.md

@@ -0,0 +1,7 @@
+## Update Azure Inventory and Configuration Connections
+
+To update a default connection to v1.9.6, execute the following SQL:
+
+~~~
+ALTER TABLE azure_collect_sql_servers_auditing_settings DROP COLUMN subscription_id;
+~~~

packs/okta.sql

@@ -0,0 +1,95 @@
+-- Users
+
+CREATE OR REPLACE VIEW data.okta_users_snapshots
+  COMMENT='all user snapshots'
+AS
+SELECT
+  event_time recorded_at,
+  raw,
+  raw:id::STRING id,
+  raw:status::STRING status,
+  raw:created::TIMESTAMP_LTZ created,
+  raw:activated::TIMESTAMP_LTZ activated,
+  raw:statusChanged::TIMESTAMP_LTZ status_changed,
+  raw:lastLogin::TIMESTAMP_LTZ last_login,
+  raw:lastUpdated::TIMESTAMP_LTZ last_updated,
+  raw:passwordChanged::TIMESTAMP_LTZ password_changed,
+  raw:profile::VARIANT profile,
+  raw:credentials::VARIANT credentials,
+  raw:_links::VARIANT links
+FROM data.okta_users_connection
+;
+
+CREATE OR REPLACE VIEW data.okta_users
+  COMMENT='latest entry seen for each user'
+AS
+SELECT *
+FROM data.okta_users_snapshots
+QUALIFY 1=ROW_NUMBER() OVER (
+  PARTITION BY id
+  ORDER BY recorded_at DESC
+)
+;
+
+-- Groups
+
+CREATE OR REPLACE VIEW data.okta_groups_snapshots
+  COMMENT='all groups snapshots'
+AS
+SELECT
+  event_time recorded_at,
+  raw,
+  raw:id::STRING id,
+  raw:created::TIMESTAMP_LTZ created,
+  raw:lastUpdated::TIMESTAMP_LTZ last_updated,
+  raw:lastMembershipUpdated::TIMESTAMP_LTZ last_membership_updated,
+  raw:objectClass::VARIANT object_class,
+  raw:type::STRING type,
+  raw:profile::VARIANT profile,
+  raw:_links::VARIANT links,
+  raw:users::VARIANT users,
+  raw:apps::VARIANT apps
+FROM data.okta_groups_connection
+;
+
+CREATE OR REPLACE VIEW data.okta_groups
+  COMMENT='latest entry seen for each group'
+AS
+SELECT *
+FROM data.okta_groups_snapshots
+QUALIFY 1=ROW_NUMBER() OVER (
+  PARTITION BY id
+  ORDER BY recorded_at DESC
+)
+;
+
+-- System Logs
+
+CREATE OR REPLACE VIEW data.okta_system_logs
+  COMMENT='all system logs'
+AS
+SELECT
+  event_time recorded_at,
+  raw,
+  raw:uuid::STRING uuid,
+  raw:published::TIMESTAMP_LTZ published,
+  raw:eventType::STRING event_type,
+  raw:version::STRING version,
+  raw:severity::STRING severity,
+  raw:legacyEventType::STRING legacy_event_type,
+  raw:displayMessage::STRING display_message,
+  raw:actor::VARIANT actor,
+  raw:client::VARIANT client,
+  raw:request::VARIANT request,
+  raw:outcome::VARIANT outcome,
+  raw:target::VARIANT target,
+  raw:transaction::VARIANT transaction,
+  raw:debugContext::VARIANT debug_context,
+  raw:authenticationContext::VARIANT authentication_context,
+  raw:securityContext::VARIANT security_context
+FROM data.okta_system_log_connection
+;
+
+SELECT * FROM data.okta_system_logs;
+SELECT * FROM data.okta_users;
+SELECT * FROM data.okta_groups;

packs/snowflake_security_monitoring.sql

@@ -97,7 +97,7 @@ WITH average_queries AS (
 )
 SELECT
   w.user_name,
-  SUM(w.query_id) AS ld_queries,
+  COUNT(w.query_id) AS ld_queries,
   a.avg_queries
 FROM snowflake.account_usage.query_history w
 JOIN average_queries a

src/connectors/aws_collect.py

@@ -8,6 +8,7 @@
     ClientError,
     DataNotFoundError,
 )
+from aiohttp.client_exceptions import ServerTimeoutError
 from collections import defaultdict, namedtuple
 import csv
 from datetime import datetime, timedelta
@@ -29,8 +30,10 @@
 READER_EID = ''
 
 _SESSION_CACHE: dict = {}
-_REQUEST_PACE_PER_SECOND = 100
-_MAX_BATCH_SIZE = 500
+
+# see https://docs.aws.amazon.com/AWSEC2/latest/APIReference/throttling.html#throttling-limits
+_REQUEST_PACE_PER_SECOND = 24  # depletes Throttling bucket of 100 at 4/s in 25s
+_REQUEST_BATCH_SIZE = 600  # 100 in Throttling bucket + 500 replenished over 25s
 
 CONNECTION_OPTIONS = [
     {
@@ -441,8 +444,8 @@
     's3_get_bucket_policy': [
         ('recorded_at', 'TIMESTAMP_LTZ'),
         ('account_id', 'STRING'),
-        ('error', 'VARIANT'),
         ('bucket', 'STRING'),
+        ('error', 'VARIANT'),
         ('policy', 'STRING'),
         ('policy_json_parsed', 'VARIANT'),
     ],
@@ -667,6 +670,22 @@
         }
     },
     'config.describe_configuration_recorders': {
+        # for unknown reasons, client.describe_regions does not seem to work w/
+        # Config client. seems like a boto3 bug. the below is a work-around.
+        'regions': [
+            'us-east-1',
+            'us-east-2',
+            'us-west-1',
+            'us-west-2',
+            'ap-south-1',
+            'ap-northeast-2',
+            'ap-southeast-2',
+            'ap-northeast-1',
+            'eu-central-1',
+            'eu-west-1',
+            'eu-west-2',
+            'eu-north-1',
+        ],
         'response': {
             'ConfigurationRecorders': [
                 {
@@ -675,7 +694,7 @@
                     'recordingGroup': 'recording_group',
                 }
             ]
-        }
+        },
     },
     'kms.list_keys': {
         'response': {'Keys': [{'KeyId': 'key_id', 'KeyArn': 'key_arn'}]},
@@ -1226,7 +1245,7 @@ async def load_task_response(client, task):
             ):
                 yield x
 
-    except (ClientError, DataNotFoundError) as e:
+    except (ClientError, DataNotFoundError, ServerTimeoutError) as e:
         log.info(format_exception_only(e))
         for x in process_aws_response(task, e):
             yield x
@@ -1240,7 +1259,11 @@ async def process_task(task, add_task) -> AsyncGenerator[Tuple[str, dict], None]
 
     try:
         now = datetime.utcnow()
-        expires = _SESSION_CACHE.get('account_arn', {}).get('Credentials', {}).get('Expiration')
+        expires = (
+            _SESSION_CACHE.get('account_arn', {})
+            .get('Credentials', {})
+            .get('Expiration')
+        )
         session = _SESSION_CACHE[account_arn] = (
             _SESSION_CACHE[account_arn]
             if expires and expires > now + timedelta(minutes=5)
@@ -1315,6 +1338,31 @@ async def aioingest(table_name, options, dryrun=False):
     AUDIT_READER_ROLE = options.get('audit_reader_role', '')
     READER_EID = options.get('reader_eid', '')
 
+    collect_apis = (
+        [
+            'iam.generate_credential_report',
+            'iam.list_account_aliases',
+            'iam.get_account_summary',
+            'iam.get_account_password_policy',
+            'ec2.describe_instances',
+            'ec2.describe_route_tables',
+            'ec2.describe_security_groups',
+            'config.describe_configuration_recorders',
+            'kms.list_keys',
+            'iam.list_users',
+            'iam.list_policies',
+            'iam.list_virtual_mfa_devices',
+            's3.list_buckets',
+            'cloudtrail.describe_trails',
+            'iam.get_credential_report',
+            'iam.list_roles',
+            'inspector.list_findings',
+            'iam.list_groups',
+        ]
+        if options.get('collect_apis', 'all') == 'all'
+        else options.get('collect_apis').split(',')
+    )
+
     oids = options.get('org_account_ids', '')
     oids = (
         [oid.strip() for oid in oids.split(',')]
@@ -1362,55 +1410,36 @@ async def aioingest(table_name, options, dryrun=False):
         )
         num_entries += len(accounts)
 
-        if options.get('collect_apis') == 'all':
-            collection_tasks = [
-                CollectTask(a['id'], method, {})
-                for method in [
-                    'iam.generate_credential_report',
-                    'iam.list_account_aliases',
-                    'iam.get_account_summary',
-                    'iam.get_account_password_policy',
-                    'ec2.describe_instances',
-                    'ec2.describe_route_tables',
-                    'ec2.describe_security_groups',
-                    'config.describe_configuration_recorders',
-                    'kms.list_keys',
-                    'iam.list_users',
-                    'iam.list_policies',
-                    'iam.list_virtual_mfa_devices',
-                    's3.list_buckets',
-                    'cloudtrail.describe_trails',
-                    'iam.get_credential_report',
-                    'iam.list_roles',
-                    'inspector.list_findings',
-                    'iam.list_groups',
-                ]
-                for a in accounts
-            ]
+        collection_tasks = [
+            CollectTask(a['id'], method, {})
+            for method in collect_apis
+            for a in accounts
+        ]
 
-            def add_task(t):
-                collection_tasks.append(t)
-
-            while collection_tasks:
-                coroutines = [
-                    aws_collect_task(
-                        t, wait=(float(i) / _REQUEST_PACE_PER_SECOND), add_task=add_task
-                    )
-                    for i, t in enumerate(collection_tasks[:_MAX_BATCH_SIZE])
-                ]
-                del collection_tasks[:_MAX_BATCH_SIZE]
-                log.info(
-                    f'progress: starting {len(coroutines)}, queued {len(collection_tasks)}'
+        def add_task(t):
+            collection_tasks.append(t)
+
+        while collection_tasks:
+            coroutines = [
+                aws_collect_task(
+                    t, wait=(i / _REQUEST_PACE_PER_SECOND), add_task=add_task
                 )
+                for i, t in enumerate(collection_tasks[:_REQUEST_BATCH_SIZE])
+            ]
+            del collection_tasks[:_REQUEST_BATCH_SIZE]
+            log.info(
+                f'progress: starting {len(coroutines)}, queued {len(collection_tasks)}'
+            )
 
-                all_results = defaultdict(list)
-                for result_lists in await asyncio.gather(*coroutines):
-                    for k, vs in result_lists.items():
-                        all_results[k] += vs
-                for name, vs in all_results.items():
-                    response = insert_list(name, vs, dryrun=dryrun)
-                    num_entries += len(vs)
-                    log.info(f'finished {name} {response}')
+            all_results = defaultdict(list)
+            for coro in asyncio.as_completed(coroutines):
+                result_lists = await coro
+                for k, vs in result_lists.items():
+                    all_results[k] += vs
+            for name, vs in all_results.items():
+                response = insert_list(name, vs, dryrun=dryrun)
+                num_entries += len(vs)
+                log.info(f'finished {name} {response}')
 
     return num_entries
 

src/connectors/azure_collect.py

@@ -716,7 +716,6 @@ def access_token_cache(cloud, client_id, tenant, secret, resource, _creds={}):
     'sql_servers_auditing_settings': [
         ('recorded_at', 'TIMESTAMP_LTZ'),
         ('tenant_id', 'VARCHAR(50)'),
-        ('subscription_id', 'VARCHAR(50)'),
         ('server_full_id', 'VARCHAR(5000)'),
         ('error', 'VARIANT'),
         ('id', 'STRING'),
@@ -1779,7 +1778,6 @@ def connect(connection_name, options):
         'response': {
             'headerDate': 'recorded_at',
             'tenantId': 'tenant_id',
-            'subscriptionId': 'subscription_id',
             'serverFullId': 'server_full_id',
             'error': 'error',
             'id': 'id',
@@ -2047,17 +2045,13 @@ def insert_results(kind, results):
 def main(
     table_name, tenant, client, secret, cloud, apis='*', dryrun=True, run_now=False
 ):
-    now = datetime.now()
-    schedule = '*' if run_now or (now.hour % 3 == 1 and now.minute < 15) else False
-
     ingest(
         table_name,
         {
             'name': 'default',
             'credentials': [
                 {'tenant': tenant, 'client': client, 'secret': secret, 'cloud': cloud}
             ],
-            'schedule': schedule,
             'apis': apis,
         },
         dryrun=dryrun,