From 31b347c340200bf7766486eaf7b7b1edd9bd8498 Mon Sep 17 00:00:00 2001 From: Ada Isopods Date: Wed, 21 May 2025 13:14:37 -0600 Subject: filter by username and password used just because --- cowrieder.py | 27 +++++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/cowrieder.py b/cowrieder.py index b5c52b7..5b7550f 100644 --- a/cowrieder.py +++ b/cowrieder.py @@ -8,6 +8,7 @@ import ipaddress from pprint import pprint cmdping_commands = ['uname', 'nproc'] # List of commands that, if used exclusively, qualify a session as a "command ping" +pubkey_attempt_str = "[SSH_PUBKEY_ATTEMPT]" # Definitions if you're reading this code for some reason: # - An empty ping is when a client connects and then does nothing, not even a log in attempt, presumably just to check if an SSH server exists @@ -35,6 +36,8 @@ parser.add_argument('-E', '--eventful', action='store_true', help='Include only parser.add_argument('-r', '--redact-ip', action='store_true', help='Redact IP addresses by converting them to 16-character hashes') parser.add_argument('-i', '--filter-ip', help='Filter data to a specific IP, censored hash thereof, or netmask (127.0.0.1 / 0b39f7265d91e7c3 / 127.0.0.0/16)') +parser.add_argument('-u', '--filter-username', help='Filter data to uses of a specific username') +parser.add_argument('-p', '--filter-password', help='Filter data to uses of a specific password. Pass "'+pubkey_attempt_str+'" to filter for attempts to log in using an SSH public key instead of a password, which does happen') parser.add_argument('-c', '--no-color', action='store_true', help="Don't color-code terminal output") @@ -143,7 +146,7 @@ def session_cred_combos(session): # List passwords used in login attempt if('password' in event): ra.append((event['username'], event['password'])) elif('key' in event): - ra.append((event['username'], "[SSH passkey attempt]")) # This means that the login attempt was made using a pubkey instead of a password + ra.append((event['username'], pubkey_attempt_str)) # This means that the login attempt was made using a pubkey instead of a password return ra def session_ran_command(session): # Was a command run during the session @@ -234,13 +237,21 @@ for session_id in sessions: if(isinstance(event[a], str)): newsessions[session_id][i][a] = event[a].replace('127.0.0.1', 'localhost').replace(src_ip, hashgen(src_ip)).replace(dst_ip, 'Host IP') - if(blankmode and ( - (not include_empty_pings and not session_attempted_login(session)) - or (not include_failed_logins and session_attempted_login(session) and not session_login_succeeded(session)) - or (not include_login_pings and session_login_succeeded(session) and not session_ran_command(session)) - or (not include_command_pings and session_is_cmdping(session)) - or (not include_eventful and session_ran_command(session) and not session_is_cmdping(session)) - )): + s_u = [] + s_p = [] + for a in session_cred_combos(session): + s_u.append(a[0]) + s_p.append(a[1]) + + if( + (blankmode and ( + (not include_empty_pings and not session_attempted_login(session)) + or (not include_failed_logins and session_attempted_login(session) and not session_login_succeeded(session)) + or (not include_login_pings and session_login_succeeded(session) and not session_ran_command(session)) + or (not include_command_pings and session_is_cmdping(session)) + or (not include_eventful and session_ran_command(session) and not session_is_cmdping(session)))) + or (args.filter_username and args.filter_username not in s_u) + or (args.filter_password and args.filter_password not in s_p)): newsessions.pop(session_id) sessions = newsessions -- cgit v1.2.3