Panther rules: impossible

from datetime import datetime, timedelta
from json import dumps, loads

import panther_event_type_helpers as event_type
from panther_base_helpers import deep_get, resolve_timestamp_string
from panther_detection_helpers.caching import get_string_set, put_string_set
from panther_ipinfo_helpers import km_between_ipinfo_loc
from panther_lookuptable_helpers import LookupTableMatches

# pylint: disable=global-variable-undefined

SATELLITE_NETWORK_ASNS = ["AS22351"]


def gen_key(event):
    """
    gen_key uses the data_model for the logtype to cache
    an entry that is specific to the Log Source ID

    The data_model needs to answer to "actor_user"
    """
    rule_name = event.get("p_source_label")
    actor = event.udm("actor_user")
    if None in [rule_name, actor]:
        return None
    return f"{rule_name.replace(' ', '')}..{actor}"


# a user-defined function that checks for client's whitelisted IP addresses
def is_ip_whitelisted(event):  # pylint: disable=unused-argument
    return False


def rule(event):
    # too-many-return-statements due to error checking
    # pylint: disable=global-statement,too-many-return-statements,too-complex,too-many-statements
    # pylint: disable=too-many-branches
    global EVENT_CITY_TRACKING
    global CACHE_KEY
    global IS_VPN
    global IS_PRIVATE_RELAY
    global IS_SATELLITE_NETWORK

    EVENT_CITY_TRACKING = {}
    CACHE_KEY = ""
    IS_VPN = False
    IS_PRIVATE_RELAY = False
    IS_SATELLITE_NETWORK = False

    # check if the IP address is in the client's whitelisted IP addresses
    if is_ip_whitelisted(event):
        return False

    # Only evaluate successful logins
    if event.udm("event_type") != event_type.SUCCESSFUL_LOGIN:
        return False

    p_event_datetime = resolve_timestamp_string(event.get("p_event_time"))
    if p_event_datetime is None:
        # we couldn't go from p_event_time to a datetime object
        # we need to do this in order to make later time comparisons generic
        return False

    new_login_stats = {
        "p_event_time": p_event_datetime.isoformat(),
        "source_ip": event.udm("source_ip"),
    }
    #
    src_ip_enrichments = LookupTableMatches().p_matches(event, event.udm("source_ip"))

    # stuff everything from ipinfo_location into the new_login_stats
    # new_login_stats is the value that we will cache for this key
    ipinfo_location = deep_get(src_ip_enrichments, "ipinfo_location")
    if ipinfo_location is None:
        return False
    new_login_stats.update(ipinfo_location)

    # Bail out if we have a None value in set as it causes false positives
    if None in new_login_stats.values():
        return False

    ## Check for VPN or Private Relay
    ipinfo_privacy = deep_get(src_ip_enrichments, "ipinfo_privacy")
    if ipinfo_privacy is not None:
        ###  Do VPN/private relay
        IS_PRIVATE_RELAY = all(
            [
                deep_get(ipinfo_privacy, "relay", default=False),
                deep_get(ipinfo_privacy, "service", default="") == "Apple Private Relay",
            ]
        )
        # We've found that some places, like WeWork locations,
        #   have the VPN attribute set to true, but do not have a
        #   service name entry.
        # We have noticed VPN connections with commercial VPN
        #   offerings have the VPN attribute set to true, and
        #   do have a service name entry
        IS_VPN = all(
            [
                deep_get(ipinfo_privacy, "vpn", default=False),
                deep_get(ipinfo_privacy, "service", default="") != "",
            ]
        )
    # Some satellite networks used during plane travel don't always
    #   register properly as VPN's, so we have a separate check here.
    IS_SATELLITE_NETWORK = (
        deep_get(src_ip_enrichments, "ipinfo_asn", "asn", default="") in SATELLITE_NETWORK_ASNS
    )

    if any((IS_VPN, IS_PRIVATE_RELAY, IS_SATELLITE_NETWORK)):
        new_login_stats.update(
            {
                "is_vpn": f"{IS_VPN}",
                "is_apple_priv_relay": f"{IS_PRIVATE_RELAY}",
                "is_satellite_network": f"{IS_SATELLITE_NETWORK}",
                "service_name": f"{deep_get(ipinfo_privacy, 'service', default='<NO_SERVICE>')}",
                "NOTE": "APPLE PRIVATE RELAY AND VPN LOGINS ARE NOT CACHED FOR COMPARISON",
            }
        )

    # Generate a unique cache key for each user per log type
    CACHE_KEY = gen_key(event)
    if not CACHE_KEY:
        # We can't save without a cache key
        return False
    # Retrieve the prior login info from the cache, if any
    last_login = get_string_set(CACHE_KEY)
    # If we haven't seen this user login in the past 1 day,
    # store this login for future use and don't alert
    if not last_login:
        if not any((IS_VPN, IS_PRIVATE_RELAY, IS_SATELLITE_NETWORK)):
            put_string_set(
                key=CACHE_KEY,
                val=[dumps(new_login_stats)],
                epoch_seconds=int((datetime.utcnow() + timedelta(days=1)).timestamp()),
            )
        return False
    # Load the last login from the cache into an object we can compare
    # str check is in place for unit test mocking
    if isinstance(last_login, str):
        tmp_last_login = loads(last_login)
        last_login = []
        for l_l in tmp_last_login:
            last_login.append(dumps(l_l))
    last_login_stats = loads(last_login.pop())

    distance = km_between_ipinfo_loc(last_login_stats, new_login_stats)
    old_time = resolve_timestamp_string(deep_get(last_login_stats, "p_event_time"))
    new_time = resolve_timestamp_string(deep_get(new_login_stats, "p_event_time"))
    time_delta = (new_time - old_time).total_seconds() / 3600  # seconds in an hour

    # Don't let time_delta be 0 (divide by zero error below)
    time_delta = time_delta or 0.0001
    # Calculate speed in Kilometers / Hour
    speed = distance / time_delta

    # Calculation is complete, write the current login to the cache
    # Only if non-VPN non-relay!
    if not any((IS_VPN, IS_PRIVATE_RELAY, IS_SATELLITE_NETWORK)):
        put_string_set(
            key=CACHE_KEY,
            val=[dumps(new_login_stats)],
            epoch_seconds=int((datetime.utcnow() + timedelta(days=1)).timestamp()),
        )

    EVENT_CITY_TRACKING["previous"] = last_login_stats
    EVENT_CITY_TRACKING["current"] = new_login_stats
    EVENT_CITY_TRACKING["speed"] = int(speed)
    EVENT_CITY_TRACKING["speed_units"] = "km/h"
    EVENT_CITY_TRACKING["distance"] = int(distance)
    EVENT_CITY_TRACKING["distance_units"] = "km"
    if deep_get(EVENT_CITY_TRACKING, "previous", "source_ip", default="<NO_PREV_IP>") == deep_get(
        EVENT_CITY_TRACKING, "current", "source_ip", default="<NO_NEW_IP>"
    ):
        # Same IP address, no alert
        return False
    return speed > 900  # Boeing 747 cruising speed


def title(event):
    #
    log_source = event.get("p_source_label", "<NO_SOURCE_LABEL>")
    old_city = deep_get(EVENT_CITY_TRACKING, "previous", "city", default="<NO_PREV_CITY>")
    new_city = deep_get(EVENT_CITY_TRACKING, "current", "city", default="<NO_PREV_CITY>")
    speed = deep_get(EVENT_CITY_TRACKING, "speed", default="<NO_SPEED>")
    distance = deep_get(EVENT_CITY_TRACKING, "distance", default="<NO_DISTANCE>")
    old_ip = deep_get(EVENT_CITY_TRACKING, "previous", "source_ip", default="<NO_PREV_IP>")
    new_ip = deep_get(EVENT_CITY_TRACKING, "current", "source_ip", default="<NO_NEW_IP>")
    return (
        f"Impossible Travel: [{event.udm('actor_user')}] "
        f"in [{log_source}] went [{speed}] km/h for [{distance}] km "
        f"between [{old_city}/{old_ip}] and [{new_city}/{new_ip}]"
    )


def dedup(event):  # pylint: disable=W0613
    return CACHE_KEY


def alert_context(event):
    context = {
        "actor_user": event.udm("actor_user"),
    }
    context.update(EVENT_CITY_TRACKING)
    return context


def severity(_):
    if any((IS_VPN, IS_PRIVATE_RELAY, IS_SATELLITE_NETWORK)):
        return "INFO"
    # time = distance/speed
    distance = deep_get(EVENT_CITY_TRACKING, "distance", default=None)
    speed = deep_get(EVENT_CITY_TRACKING, "speed", default=None)
    if speed and distance:
        time = distance / speed
        # time of 0.1666 is 10 minutes
        if time < 0.1666 and distance < 50:
            # This is likely a GEOIP inaccuracy
            return "LOW"
    return "HIGH"

AnalysisType: rule
Filename: impossible_travel_login.py
RuleID: "Standard.ImpossibleTravel.Login"
DisplayName: "Impossible Travel for Login Action"
Enabled: true
LogTypes:
  - Asana.Audit
  - AWS.CloudTrail
  - Notion.AuditLogs
  - Okta.SystemLog
Tags:
  - Identity & Access Management
  - Initial Access:Valid Accounts
Reports:
  MITRE ATT&CK:
    - TA0001:T1078
Severity: High
Description: A user has subsequent logins from two geographic locations that are very far apart
Runbook: |
  Reach out to the user if needed to validate the activity, then lock the account.

  If the user responds that the geolocation on the new location is incorrect, you can directly
  report the inaccuracy via  https://ipinfo.io/corrections
Reference: https://expertinsights.com/insights/what-are-impossible-travel-logins/#:~:text=An%20impossible%20travel%20login%20is,of%20the%20logins%20is%20fraudulent
SummaryAttributes:
  - p_any_usernames
  - p_any_ip_addresses
  - p_any_domain_names
# All test cases for this detection will need to include:
# * p_log_type ( for udm)
# Alerting test cases for this detection will need to include:
# * p_source_label
# * p_event_time