Panther rules: teleport

from datetime import datetime, timedelta
from typing import Dict, Tuple

from panther_base_helpers import (
    golang_nanotime_to_python_datetime,
    panther_nanotime_to_python_datetime,
)

PANTHER_TIME_FORMAT = r"%Y-%m-%d %H:%M:%S.%f"
# Tune this to be some Greatest Common Denominator of session TTLs for your
# environment
MAXIMUM_NORMAL_VALIDITY_INTERVAL = timedelta(hours=12)
# To allow some time in between when a request is submitted and authorized
# vs when the certificate actually gets generated. In practice, this is much
# less than 5 seconds.
ISSUANCE_GRACE_PERIOD = timedelta(seconds=5)

# You can audit your logs in Panther to try and understand your role/validity
# patterns from a known-good period of access.
# A query example:
# ```sql
#  SELECT
#     cluster_name,
#     identity:roles,
#     DATEDIFF('HOUR', time, identity:expires) AS validity
#  FROM
#     panther_logs.public.gravitational_teleportaudit
#  WHERE
#     p_occurs_between('2023-09-01 00:00:00','2023-10-06 21:00:00Z')
#     AND event = 'cert.create'
#  GROUP BY cluster_name, identity:roles, validity
#  ORDER BY validity DESC
# ```

# A dictionary of:
#  cluster names: to a dictionary of:
#     role names: mapping to a tuple of:
#        ( maximum usual validity, expiration datetime for this rule )
CLUSTER_ROLE_MAX_VALIDITIES: Dict[str, Dict[str, Tuple[timedelta, datetime]]] = {
    # "teleport.example.com": {
    #     "example_role": (timedelta(hours=720), datetime(2023, 12, 01, 01, 02, 03)),
    #     "other_example_role": (timedelta(hours=720), datetime.max),
    # },
}


def rule(event):
    if not event.get("event") == "cert.create":
        return False
    max_validity = MAXIMUM_NORMAL_VALIDITY_INTERVAL + ISSUANCE_GRACE_PERIOD
    for role in event.deep_get("identity", "roles", default=[]):
        validity, expiration = CLUSTER_ROLE_MAX_VALIDITIES.get(event.get("cluster_name"), {}).get(
            role, (None, None)
        )
        if validity and expiration:
            # Ignore exceptions that have passed their expiry date
            if datetime.utcnow() < expiration:
                max_validity = max(max_validity, validity)
    return validity_interval(event) > max_validity


def validity_interval(event):
    event_time = panther_nanotime_to_python_datetime(event.get("time"))
    expires = golang_nanotime_to_python_datetime(
        event.deep_get("identity", "expires", default=None)
    )
    if not event_time and expires:
        return False
    interval = expires - event_time
    return interval


def title(event):
    identity = event.deep_get("identity", "user", default="<Cert with no User!?>")
    return (
        f"A Certificate for [{identity}] "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}] "
        f"has been issued for an unusually long time: {validity_interval(event)!r} "
    )

AnalysisType: rule
Filename: teleport_long_lived_certs.py
RuleID: Teleport.LongLivedCerts
DisplayName: A long-lived cert was created
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - Teleport
Severity: Medium
Description: An unusually long-lived Teleport certificate was created
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0003:T1098
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  Teleport certificates are usually issued for a short period of time. Alert if long-lived certificates were created.
SummaryAttributes:
  - event
  - code
  - time
  - identity

def rule(event):
    return event.get("event") == "saml.created"


def title(event):
    return (
        f"A SAML connector was created or updated by User [{event.get('user', '<UNKNOWN_USER>')}] "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_saml_created.py
RuleID: Teleport.SAMLCreated
DisplayName: A SAML Connector was created or modified
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - Teleport
Severity: High
Description: A SAML connector was created or modified
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0042:T1585
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  When a SAML connector is modified, it can potentially change the trust model of the Teleport Cluster. Validate that these changes were expected and correct.
SummaryAttributes:
  - event
  - code
  - user
  - name

def rule(event):
    return event.get("event") == "lock.created"


def title(event):
    return (
        f"A Teleport Lock was created by {event.get('updated_by', '<UNKNOWN_UPDATED_BY>')} "
        f"to Lock out user {event.get('target', {}).get('user', '<UNKNOWN_USER>')} "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_lock_created.py
RuleID: Teleport.LockCreated
DisplayName: A Teleport Lock was created
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - Teleport
Severity: Info
Description: A Teleport Lock was created
DedupPeriodMinutes: 60
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  A Teleport Lock was created; this is an unusual administrative action. Investigate to understand why a Lock was created.
SummaryAttributes:
  - event
  - code
  - time
  - identity

def rule(event):
    return event.get("event") == "role.created"


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] created Role "
        f"[{event.get('name', '<UNKNOWN_NAME>')}] "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_role_created.py
RuleID: Teleport.RoleCreated
DisplayName: A Teleport Role was modified or created
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - Teleport
Severity: Medium
Description: A Teleport Role was modified or created
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0003:T1098.001
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  A Teleport Role was modified or created. Validate its legitimacy.
SummaryAttributes:
  - event
  - code
  - user
  - name

def rule(event):
    cluster = event.get("cluster_name", "")
    user_domain = event.get("user", "@").split("@")[-1]
    return (
        event.get("event") == "user.login"
        and event.get("success") is True
        and event.get("method") == "saml"
        and not cluster.endswith(user_domain)
    )


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] logged into "
        f"[{event.get('cluster_name', '<UNNAMED_CLUSTER>')}] using SAML from a different domain"
    )

AnalysisType: rule
Filename: teleport_saml_login_not_company_domain.py
RuleID: Teleport.SAMLLoginWithoutCompanyDomain
DisplayName: "A user authenticated with SAML, but from an unknown company domain"
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - Teleport
Severity: Medium
Description: "A user authenticated with SAML, but from an unknown company domain"
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0003:T1098
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  A user authenticated with SAML, but from an unknown company domain
SummaryAttributes:
  - event
  - code
  - user
  - method
  - mfa_device

def rule(event):
    user_domain = event.get("user", "@").split("@")[-1]
    cluster = event.get("cluster_name", "")
    return bool(
        event.get("event") == "user.login"
        and event.get("success") is True
        and cluster.endswith(user_domain)
        and event.get("method") != "saml"
    )


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] logged into "
        f"[{event.get('cluster_name', '<UNNAMED_CLUSTER>')}] without using SAML"
    )

AnalysisType: rule
Filename: teleport_company_domain_login_without_saml.py
RuleID: Teleport.CompanyDomainLoginWithoutSAML
DisplayName: "A User from the company domain(s) Logged in without SAML"
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - Teleport
Severity: Medium
Description: "A User from the company domain(s) Logged in without SAML"
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0005:T1562
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  A User from the company domain(s) Logged in without SAML
SummaryAttributes:
  - event
  - code
  - user
  - method
  - mfa_device

from panther_base_helpers import pattern_match_list

USER_CREATE_PATTERNS = [
    "chage",  # user password expiry
    "passwd",  # change passwords for users
    "user*",  # create, modify, and delete users
]


def rule(event):
    # Filter the events
    if event.get("event") != "session.command":
        return False
    # Check that the program matches our list above
    return pattern_match_list(event.get("program", ""), USER_CREATE_PATTERNS)


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] has manually modified system users "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_create_user_accounts.py
RuleID: "Teleport.CreateUserAccounts"
DisplayName: "Teleport Create User Accounts"
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - SSH
  - Persistence:Create Account
Reports:
  MITRE ATT&CK:
    - TA0003:T1136
Severity: High
Description: A user has been manually created, modified, or deleted
DedupPeriodMinutes: 15
Reference: https://goteleport.com/docs/management/admin/
Runbook: Analyze why it was manually created and delete it if necessary.
SummaryAttributes:
  - event
  - code
  - user
  - program
  - path
  - return_code
  - login
  - server_id
  - sid

SCAN_COMMANDS = {"arp", "arp-scan", "fping", "nmap"}


def rule(event):
    # Filter out commands
    if event.get("event") == "session.command" and not event.get("argv"):
        return False
    # Check that the program is in our watch list
    return event.get("program") in SCAN_COMMANDS


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] has issued a network scan with "
        f"[{event.get('program', '<UNKNOWN_PROGRAM>')}] "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_network_scanning.py
RuleID: "Teleport.NetworkScanning"
DisplayName: "Teleport Network Scan Initiated"
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - SSH
  - Discovery:Network Service Discovery
Severity: Medium
Description: A user has invoked a network scan that could potentially indicate enumeration of the network.
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0007:T1046
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  Find related commands within the time window and determine if the command was invoked legitimately. Examine the arguments to determine how the command was used.
SummaryAttributes:
  - event
  - code
  - user
  - program
  - path
  - return_code
  - login
  - server_id
  - sid

def rule(event):
    # Filter the events
    if event.get("event") != "session.command":
        return False
    # Ignore list/read events
    if "-l" in event.get("argv", []):
        return False
    return event.get("program") == "crontab"


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] has modified scheduled jobs"
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_scheduled_jobs.py
RuleID: "Teleport.ScheduledJobs"
DisplayName: "Teleport Scheduled Jobs"
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - SSH
  - Execution:Scheduled Task/Job
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0002:T1053
Description: A user has manually edited the Linux crontab
Threshold: 10
DedupPeriodMinutes: 15
Reference: https://goteleport.com/docs/management/admin/
Runbook: Validate the user behavior and rotate the host if necessary.
SummaryAttributes:
  - event
  - code
  - user
  - program
  - path
  - return_code
  - login
  - server_id
  - sid

def rule(event):
    return bool(event.get("error")) and event.get("event") == "auth"


def title(event):
    return (
        f"A high volume of SSH errors was detected from user "
        f"[{event.get('user', '<UNKNOWN_USER>')}] "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_auth_errors.py
RuleID: "Teleport.AuthErrors"
DisplayName: "Teleport SSH Auth Errors"
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - SSH
  - Credential Access:Brute Force
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0006:T1110
Description: A high volume of SSH errors could indicate a brute-force attack
Threshold: 10
DedupPeriodMinutes: 15
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  Check that the user making the failed requests legitimately tried logging in that many times.
SummaryAttributes:
  - event
  - code
  - user
  - program
  - path
  - return_code
  - login
  - server_id
  - sid

SUSPICIOUS_COMMANDS = {"nc", "wget"}


def rule(event):
    if event.get("event") != "session.command":
        return False
    # Ignore commands without arguments
    if not event.get("argv"):
        return False
    return event.get("program") in SUSPICIOUS_COMMANDS


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] has executed the command "
        f"[{event.get('program', '<UNKNOWN_PROGRAM>')}] "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_suspicious_commands.py
RuleID: "Teleport.SuspiciousCommands"
DisplayName: "Teleport Suspicious Commands Executed"
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - SSH
  - Execution:Command and Scripting Interpreter
Severity: Medium
Description: A user has invoked a suspicious command that could lead to a host compromise
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0002:T1059
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  Find related commands within the time window and determine if the command was invoked legitimately. Examine the arguments to determine how the command was used and reach out to the user to verify the intentions.
SummaryAttributes:
  - event
  - code
  - user
  - program
  - path
  - return_code
  - login
  - server_id
  - sid

def rule(event):
    return event.get("event") == "session.start" and event.get("login") == "root"


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] logged into "
        f"[{event.get('server_hostname', '<UNKNOWN_HOSTNAME>')}] as root "
        f"on [{event.get('cluster_name', '<UNKNOWN_CLUSTER>')}]"
    )

AnalysisType: rule
Filename: teleport_root_login.py
RuleID: Teleport.RootLogin
DisplayName: User Logged in as root
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - SSH
  - Execution:Command and Scripting Interpreter
  - Teleport
Severity: Medium
Description: A User logged in as root
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0002:T1059
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  Use user accounts and policies, rather than root when logging in. With access to root, it is possible to evade auditing and logging.
SummaryAttributes:
  - event
  - code
  - user
  - login
  - server_hostname
  - server_labels

SENSITIVE_LOCAL_USERS = ["breakglass"]


def rule(event):
    return (
        event.get("event") == "user.login"
        and event.get("success") == "true"
        and event.get("method") == "local"
        and not event.get("mfa_device")
    )


def severity(event):
    if event.get("user") in SENSITIVE_LOCAL_USERS:
        return "HIGH"
    return "MEDIUM"


def title(event):
    return (
        f"User [{event.get('user', '<UNKNOWN_USER>')}] logged into "
        f"[{event.get('cluster_name', '<UNNAMED_CLUSTER>')}] locally "
        f"without using MFA"
    )

AnalysisType: rule
Filename: teleport_local_user_login_without_mfa.py
RuleID: Teleport.LocalUserLoginWithoutMFA
DisplayName: User Logged in wihout MFA
Enabled: true
LogTypes:
  - Gravitational.TeleportAudit
Tags:
  - Teleport
Severity: High
Description: A local User logged in without MFA
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0001:T1078
Reference: https://goteleport.com/docs/management/admin/
Runbook: >
  A local user logged in without Multi-Factor Authentication
SummaryAttributes:
  - event
  - code
  - user
  - success
  - mfa_device