oss helper reorg

global_helpers/default_test.py

@@ -6,11 +6,11 @@
 import unittest
 
 sys.path.append(os.path.dirname(__file__))
-import panther_aws_helpers as p_d  # pylint: disable=C0413
+import panther_aws_helpers as p_aws_h  # pylint: disable=C0413
 
 
 class TestAWSKeyAccountId(unittest.TestCase):
     def test_aws_key_account_id(self):
         aws_key_id = "ASIAY34FZKBOKMUTVV7A"
-        account_id = p_d.aws_key_account_id(aws_key_id)
+        account_id = p_aws_h.aws_key_account_id(aws_key_id)
         self.assertEqual(account_id, "609629065308")

global_helpers/panther_aws_helpers.py

@@ -1,10 +1,24 @@
 import base64
 import binascii
-from typing import List
+import os
+from typing import Any, Dict, List
 
+import boto3
 from panther_config import config
 
+
+class BadLookup(Exception):
+    """Error returned when a resource lookup fails."""
+
+
+class PantherBadInput(Exception):
+    """Error returned when a Panther helper function is provided bad input."""
+
+
 AWS_ACCOUNTS = config.AWS_ACCOUNTS
+_RESOURCE_TABLE = None  # boto3.Table resource, lazily constructed
+FIPS_ENABLED = os.getenv("ENABLE_FIPS", "").lower() == "true"
+FIPS_SUFFIX = "-fips." + os.getenv("AWS_REGION", "") + ".amazonaws.com"
 
 
 def aws_strip_role_session_id(user_identity_arn):
@@ -162,3 +176,50 @@ def lookup_aws_account_name(account_id):
         str: The AWS account ID (unnamed account)
     """
     return AWS_ACCOUNTS.get(account_id, f"{account_id} (unnamed account)")
+
+
+def get_s3_arn_by_name(name: str) -> str:
+    """This function is used to construct an s3 bucket ARN from its name."""
+    if name == "":
+        raise PantherBadInput("s3 name cannot be blank")
+    return "arn:aws:s3:::" + name
+
+
+def s3_lookup_by_name(name: str) -> Dict[str, Any]:
+    """This function is used to get an S3 bucket resource from just its name."""
+    return resource_lookup(get_s3_arn_by_name(name))
+
+
+def resource_table() -> boto3.resource:
+    """Lazily build resource table"""
+    # pylint: disable=global-statement
+    global _RESOURCE_TABLE
+    if not _RESOURCE_TABLE:
+        # pylint: disable=no-member
+        _RESOURCE_TABLE = boto3.resource(
+            "dynamodb",
+            endpoint_url="https://dynamodb" + FIPS_SUFFIX if FIPS_ENABLED else None,
+        ).Table("panther-resources")
+    return _RESOURCE_TABLE
+
+
+def resource_lookup(resource_id: str) -> Dict[str, Any]:
+    """This function is used to get a resource from the resources-api based on its resourceID."""
+    # Validate input so we can provide meaningful error messages to users
+    if resource_id == "":
+        raise PantherBadInput("resourceId cannot be blank")
+
+    # Get the item from dynamo
+    response = resource_table().get_item(Key={"id": resource_id})
+
+    # Check if dynamo failed
+    status_code = response["ResponseMetadata"]["HTTPStatusCode"]
+    if status_code != 200:
+        raise BadLookup("dynamodb - " + str(status_code) + " HTTPStatusCode")
+
+    # Check if the item was found
+    if "Item" not in response:
+        raise BadLookup(resource_id + " not found")
+
+    # Return just the attributes of the item
+    return response["Item"]["attributes"]

global_helpers/panther_base_helpers.py

@@ -9,6 +9,8 @@
 from ipaddress import ip_address, ip_network
 from typing import Any, List, Optional, Sequence, Union
 
+from dateutil import parser
+
 # # # # # # # # # # # # # #
 #       Exceptions        #
 # # # # # # # # # # # # # #
@@ -198,3 +200,130 @@ def listify(maybe_list):
         return [maybe_list]
     # either a list or string
     return [maybe_list] if isinstance(maybe_list, (str, bytes, dict)) else maybe_list
+
+
+# Auto Time Resolution Parameters
+EPOCH_REGEX = r"([0-9]{9,12}(\.\d+)?)"
+TIME_FORMATS = [
+    "%Y-%m-%d %H:%M:%S",  # Panther p_event_time Timestamp
+    "%Y-%m-%dT%H:%M:%SZ",  # AWS Timestamp
+    "%Y-%m-%dT%H:%M:%S.%fZ",  # Panther Timestamp
+    "%Y-%m-%dT%H:%M:%S*%f%z",
+    "%Y %b %d %H:%M:%S.%f %Z",
+    "%b %d %H:%M:%S %z %Y",
+    "%d/%b/%Y:%H:%M:%S %z",
+    "%b %d, %Y %I:%M:%S %p",
+    "%b %d %Y %H:%M:%S",
+    "%b %d %H:%M:%S %Y",
+    "%b %d %H:%M:%S %z",
+    "%b %d %H:%M:%S",
+    "%Y-%m-%dT%H:%M:%S%z",
+    "%Y-%m-%dT%H:%M:%S.%f%z",
+    "%Y-%m-%d %H:%M:%S %z",
+    "%Y-%m-%d %H:%M:%S%z",
+    "%Y-%m-%d %H:%M:%S,%f",
+    "%Y/%m/%d*%H:%M:%S",
+    "%Y %b %d %H:%M:%S.%f*%Z",
+    "%Y %b %d %H:%M:%S.%f",
+    "%Y-%m-%d %H:%M:%S,%f%z",
+    "%Y-%m-%d %H:%M:%S.%f",
+    "%Y-%m-%d %H:%M:%S.%f%z",
+    "%Y-%m-%dT%H:%M:%S.%f",
+    "%Y-%m-%dT%H:%M:%S",
+    "%Y-%m-%dT%H:%M:%S%z",
+    "%Y-%m-%dT%H:%M:%S.%f",
+    "%Y-%m-%dT%H:%M:%S",
+    "%Y-%m-%d*%H:%M:%S:%f",
+    "%Y-%m-%d*%H:%M:%S",
+    "%y-%m-%d %H:%M:%S,%f %z",
+    "%y-%m-%d %H:%M:%S,%f",
+    "%y-%m-%d %H:%M:%S",
+    "%y/%m/%d %H:%M:%S",
+    "%y%m%d %H:%M:%S",
+    "%Y%m%d %H:%M:%S.%f",
+    "%m/%d/%y*%H:%M:%S",
+    "%m/%d/%Y*%H:%M:%S",
+    "%m/%d/%Y*%H:%M:%S*%f",
+    "%m/%d/%y %H:%M:%S %z",
+    "%m/%d/%Y %H:%M:%S %z",
+    "%H:%M:%S",
+    "%H:%M:%S.%f",
+    "%H:%M:%S,%f",
+    "%d/%b %H:%M:%S,%f",
+    "%d/%b/%Y:%H:%M:%S",
+    "%d/%b/%Y %H:%M:%S",
+    "%d-%b-%Y %H:%M:%S",
+    "%d-%b-%Y %H:%M:%S.%f",
+    "%d %b %Y %H:%M:%S",
+    "%d %b %Y %H:%M:%S*%f",
+    "%m%d_%H:%M:%S",
+    "%m%d_%H:%M:%S.%f",
+    "%m/%d/%Y %I:%M:%S %p:%f",
+    "%m/%d/%Y %I:%M:%S %p",
+]
+
+
+def resolve_timestamp_string(timestamp: str) -> Optional[datetime]:
+    """Auto Time Resolution"""
+    if not timestamp:
+        return None
+
+    # Removes weird single-quotes used in some timestamp formats
+    ts_format = timestamp.replace("'", "")
+    # Attempt to resolve timestamp format
+    for each_format in TIME_FORMATS:
+        try:
+            return datetime.strptime(ts_format, each_format)
+        except (ValueError, TypeError):
+            continue
+    try:
+        return parser.parse(timestamp)
+    except (ValueError, TypeError, parser.ParserError):
+        pass
+
+    # Attempt to resolve epoch format
+    # Since datetime.utcfromtimestamp supports 9 through 12 digit epoch timestamps
+    # and we only want the first 12 digits.
+    match = re.match(EPOCH_REGEX, timestamp)
+    if match.group(0) != "":
+        try:
+            return datetime.utcfromtimestamp(float(match.group(0)))
+        except (ValueError, TypeError):
+            return None
+    return None
+
+
+# returns the difference between time1 and later time 2 in human-readable time period string
+def time_delta(time1, time2: str) -> str:
+    time1_truncated = nano_to_micro(time1)
+    time2_truncated = nano_to_micro(time2)
+    delta_timedelta = resolve_timestamp_string(time2_truncated) - resolve_timestamp_string(
+        time1_truncated
+    )
+    days = delta_timedelta.days
+    hours, remainder = divmod(delta_timedelta.seconds, 3600)
+    minutes, seconds = divmod(remainder, 60)
+    delta = ""
+    if days > 0:
+        delta = f"{days} day(s) "
+    if hours > 0:
+        delta = "".join([delta, f"{hours} hour(s) "])
+    if minutes > 0:
+        delta = "".join([delta, f"{minutes} minute(s) "])
+    if seconds > 0:
+        delta = "".join([delta, f"{seconds} second(s)"])
+    return delta
+
+
+def nano_to_micro(time_str: str) -> str:
+    parts = time_str.split(":")
+    # pylint: disable=consider-using-f-string
+    parts[-1] = "{:06f}".format(float(parts[-1]))
+    return ":".join(parts)
+
+
+# adds parsing delay to an alert_context
+def add_parse_delay(event, context: dict) -> dict:
+    parsing_delay = time_delta(event.get("p_event_time"), event.get("p_parse_time"))
+    context["parseDelay"] = f"{parsing_delay}"
+    return context

global_helpers/panther_gcp_helpers.py

@@ -59,7 +59,6 @@ def get_binding_deltas(event):
     if not service_data:
         return []
 
-    # Reference: bit.ly/2WsJdZS
     binding_deltas = service_data.get("policyDelta", {}).get("bindingDeltas")
     if not binding_deltas:
         return []

global_helpers/panther_ipinfo_helpers.py

@@ -180,7 +180,7 @@ def geoinfo_from_ip(event, match_field: str):
     }
 
 
-def geoinfo_from_ip_formatted(event, match_field: str) -> str:  # pylint: disable=invalid-name
+def geoinfo_from_ip_formatted(event, match_field: str) -> str:
     """Formatting wrapper for geoinfo_from_ip for use in human-readable text"""
     geoinfo = geoinfo_from_ip(event, match_field)
     return (