From 8fd7d671c58f7c3b84b0c2ba5fe628742c8c2590 Mon Sep 17 00:00:00 2001 From: Ada Isopods Date: Sat, 17 May 2025 05:08:56 -0600 Subject: ip stats --- cowrieder.py | 247 +++++++++++++++++++++++++++++++++++++---------------------- 1 file changed, 154 insertions(+), 93 deletions(-) (limited to 'cowrieder.py') diff --git a/cowrieder.py b/cowrieder.py index 01d93fe..f18d230 100644 --- a/cowrieder.py +++ b/cowrieder.py @@ -5,6 +5,40 @@ import os import argparse import hashlib +cmdping_commands = ['uname', 'nproc'] # List of commands that, if used exclusively, qualify a session as a "command ping" + +# 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 +# - A login ping is when a client logs in successfully and leaves without doing anything +# - A "command ping" is when a client logs in and then only runs the handful of common test/query commands listed above. These make up such a +# large percentage of sessions where any command is run that I felt the need to create a distinction in order to make finding the most interesting log entires easier + +# ArgParse crap coming in hot + +parser = argparse.ArgumentParser( + description="Script to parse JSON logs from popular SSH honeypot Cowrie and print them in a more readable format" +) + +parser.add_argument('filepath') + +parser.add_argument('-v', '--verbose', action='store_true', help="Make the output more verbose (print abridged logs for all logins where commands are run, not just the interesting ones)") +parser.add_argument('-V', '--verboser', action='store_true', help="Make the output even more verbose (print timestamp, event id, and message (if applicable) for all events)") +parser.add_argument('-W', '--verbosest', action='store_true', help="Make the output even more verboserer (just group the input log by session id and dump it to pretty-printed JSON)") + +parser.add_argument('-P', '--pings', action='store_true', help='Include only connections where no authentication attempt was made') +parser.add_argument('-F', '--fails', action='store_true', help='Include only failed logins') +parser.add_argument('-L', '--logins', action='store_true', help='Include only successful logins where no commands were run') +parser.add_argument('-C', '--commands', action='store_true', help='Include only logins where only this list of basic test commands were run: ' + str(cmdping_commands)) +parser.add_argument('-E', '--eventful', action='store_true', help='Include only the cool interesting sessions that aren\'t any of that other stuff') + +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 or censored hash thereof') + +parser.add_argument('-c', '--no-color', action='store_true', help="Don't color-code terminal output") + +parser.add_argument('-s', '--stats', action='store_true', help='Print aggregated statistics') +parser.add_argument('-S', '--only-stats', action='store_true', help='Print only aggregated statistics') + def ccgen(seed=""): # Color code generator if(not os.isatty(1)): # Do not output color code at all if output is not a TTY that can accept them return "" @@ -49,35 +83,7 @@ def hashgen(seed, length=16): raise Warning("Due to strange behaviour on the part of the hexdigest() function, length should be a multiple of 2") return hashlib.shake_256(str(seed).encode("utf-8")).hexdigest(int(length / 2)) -cmdping_commands = ['uname', 'nproc', 'echo'] # List of commands that, if used exclusively, qualify a session as a "command ping" - -# ArgParse crap coming in hot - -parser = argparse.ArgumentParser( - description="Script to parse JSON logs from popular SSH honeypot Cowrie and print them in a more readable format" -) - -parser.add_argument('filepath') - -parser.add_argument('-v', '--verbose', action='store_true', help="Make the output more verbose (print abridged logs for all logins where commands are run, not just the interesting ones)") -parser.add_argument('-V', '--verboser', action='store_true', help="Make the output even more verbose (print timestamp, event id, and message (if applicable) for all events)") -parser.add_argument('-W', '--verbosest', action='store_true', help="Make the output even more verboserer (just group the input log by session id and dump it to pretty-printed JSON)") - -parser.add_argument('-P', '--pings', action='store_true', help='Include only connections where no authentication attempt was made') -parser.add_argument('-F', '--fails', action='store_true', help='Include only failed logins') -parser.add_argument('-L', '--logins', action='store_true', help='Include only successful logins where no commands were run') -parser.add_argument('-C', '--commands', action='store_true', help='Include only logins where only this list of basic test commands were run: ' + str(cmdping_commands)) -parser.add_argument('-E', '--eventful', action='store_true', help='Include only the cool interesting sessions that aren\'t any of that other stuff') - -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 or censored hash thereof') - -parser.add_argument('-s', '--stats', action='store_true', help='Print aggregated statistics') -parser.add_argument('-S', '--only-stats', action='store_true', help='Print only aggregated statistics') - - args = parser.parse_args() -print_stats = args.stats or args.only_stats verbosity = 0 if(args.verbosest): @@ -89,7 +95,7 @@ elif(args.verbose): filepath = args.filepath -blankmode = (args.pings or args.fails or args.logins or args.commands or args.eventful or args.only_stats) +blankmode = (args.pings or args.fails or args.logins or args.commands or args.eventful) include_empty_pings = (not blankmode or args.pings) include_failed_logins = (not blankmode or args.fails) @@ -97,6 +103,8 @@ include_login_pings = (not blankmode or args.logins) include_command_pings = (not blankmode or args.commands) include_eventful = (not blankmode or args.eventful) +print_stats = args.stats or args.only_stats + sessions = {} # Dict - session_id : list of events in session def session_ip(session): # Grab IP from session @@ -224,72 +232,125 @@ for session_id in sessions: sessions = newsessions +# Statistics Aggregation + +attempt_counts_by_ip = {} +ips_by_attempt_count = {} +def top_n_percent_of_ips_session_percent(n): # pretty self explanatory. Pass a float between 0 and 1 + i = 0 + maxi = len(attempt_counts_by_ip) * n + c = 0 + for ip in attempt_counts_by_ip: + if(i > maxi): + break + c += attempt_counts_by_ip[ip] + i += 1 + return int(c / len(sessions) * 100) + + +for s_id in sessions: + s = sessions[s_id] + ip = session_ip(s) + if(ip not in attempt_counts_by_ip): + attempt_counts_by_ip.update({ip: 1}) + else: + attempt_counts_by_ip[ip] += 1 +attempt_counts_by_ip = dict(sorted(attempt_counts_by_ip.items(), key=lambda item: item[1], reverse=True)) +for ip in attempt_counts_by_ip: + if(attempt_counts_by_ip[ip] not in ips_by_attempt_count): + ips_by_attempt_count.update({attempt_counts_by_ip[ip]: []}) + ips_by_attempt_count[attempt_counts_by_ip[ip]].append(ip) + +def printstats(): + print("Source IP addresses by attempt count:") + for num in ips_by_attempt_count: + out = (" " * (len(str(next(iter(ips_by_attempt_count)))) - len(str(num)))) + str(num) + ": " + ipcount = 0 + for ip in ips_by_attempt_count[num]: + if(verbosity == 0 and ipcount >= 3): + out += str(len(ips_by_attempt_count[num])) + " more" + break + out += ppip(ip, ttycolor=True) + if(ipcount < len(ips_by_attempt_count[num]) - 1): + out += ", " + ipcount += 1 + print(out) + print("Of a total " + str(len(sessions)) + " sessions:") + print("The top 50% of IPs were responsible for " + str(top_n_percent_of_ips_session_percent(0.5)) + "%") + print("The top 10% of IPs were responsible for " + str(top_n_percent_of_ips_session_percent(0.1)) + "%") + print("The top 1% of IPs were responsible for " + str(top_n_percent_of_ips_session_percent(0.01)) + "%") + print("The top single IP was responsible for " + str(int(attempt_counts_by_ip[next(iter(attempt_counts_by_ip))] / len(sessions) * 100)) + "%") + # Printing logic coming up -if(verbosity == 2): # High verbosity mode: print timestamp, event type and message (if applicable) for every event. I could probably do better but I'm lazy - for session_id in sessions: - session = sessions[session_id] - ip = session_ip(session_id) - print("\n--- Session " + ccgen(session_id) + session_id + ccgen() + " from " + ppip(ip, ttycolor=True) + " to " + session_sensor(session) + " ---") - for entry in session: - eventid = entry['eventid'] - if(entry['eventid'] == "cowrie.session.params"): - print(entry["timestamp"] + " > " + eventid + (" " * (29 - len(eventid))) + ": ARCH: " + entry["arch"]) - else: - print(entry["timestamp"] + " > " + eventid + (" " * (29 - len(eventid))) + ": " + entry['message']) - -elif(verbosity == 3): # Maximum verbosity: literally just dump the now-sorted-by-session logs as pretty-printed JSON +if(verbosity == 3): # Maximum verbosity: literally just dump the now-sorted-by-session logs as pretty-printed JSON print(json.dumps(sessions, indent=4)) + exit() + +elif(verbosity == 2): # High verbosity mode: print timestamp, event type and message (if applicable) for every event. I could probably do better but I'm lazy + if(not args.only_stats): + for session_id in sessions: + session = sessions[session_id] + ip = session_ip(session) + print("\n--- Session " + ccgen(session_id) + session_id + ccgen() + " from " + ppip(ip, ttycolor=True) + " to " + session_sensor(session) + " ---") + for entry in session: + eventid = entry['eventid'] + if(entry['eventid'] == "cowrie.session.params"): + print(entry["timestamp"] + " > " + eventid + (" " * (29 - len(eventid))) + ": ARCH: " + entry["arch"]) + else: + print(entry["timestamp"] + " > " + eventid + (" " * (29 - len(eventid))) + ": " + entry['message']) + if(print_stats): + print("TODO -- this shit's gonna be verbose as fuck man") else: # Default (low) verbosity: just the good stuff - for session_id in sessions: - - # Declaring initial variables - session = sessions[session_id] - ip = session_ip(session) - timestamp = session_timestamp(session)[:22].replace("T", " ") # clean up the timestamp text all pretty like - durationstring = "closed after " + str(session_duration(session)) + "s" - if(session_duration(session) < 0): - durationstring = "active as of processing" - - if(not session_attempted_login(session)): # Printing empty pings - print(timestamp + " > " + ppip(ip, True, True) + " > Connection opened and " + durationstring) - - elif(session_attempted_login(session) and not session_login_succeeded(session)): # Printing failed logins - print(timestamp + " > " + ppip(ip, True, True) + " > " + ccgen("RED") +"Failed login" + ccgen() + " with credentials: " + str(session_cred_combos(session))) - - elif(session_ran_command(session)): # Printing logic for sessions in which a command was run - - if(not verbosity == 1 and session_is_cmdping(session)): # If -v is not enabled, collapse "command pings" to one-liners - print(timestamp + " > " + ppip(ip, True, True) + " > " + ccgen("GREEN") + "Login" + ccgen() + " with credentials: " + str(session_cred_combos(session)) + ccgen("BLUE") + " ran commands " + ccgen() + str(session_pingcmds(session))) - - else: # Printing logic for un-collapsed sessions where commands were run - print(timestamp + " > " + ppip(ip, True, True) + " > " + ccgen("CYAN") + "Session" + ccgen() + ":") - for entry in session: - match(entry["eventid"]): - case "cowrie.login.failed": - pw = [] - if("password" in entry): - pw = entry["password"] - print(" | Failed login: " + entry["username"] + "/" + str(pw)) - case "cowrie.login.success": - print(" | Successful login: " + entry["username"] + "/" + entry["password"]) - case "cowrie.command.input": - print(" | CMD: " + entry["input"]) - case "cowrie.session.file_download": - df = "" - if("url" in entry): - df = entry["url"] - elif("destfile" in entry): - df = entry["destfile"] - if(verbosity == 1): - print(" | Saved file " + df + " to " + entry["outfile"]) - else: - print(" | Saved file " + df + " to " + entry["outfile"][:40] + "...") - print(" X Session " + durationstring) - - else: # Printing logic for successful logins in which no commands were run - print(timestamp + " > " + ppip(ip, True, True) + " > " + ccgen("GREEN") + "Login" + ccgen() + " with credentials: " + str(session_cred_combos(session))) - -if(args.stats): - print("Aggregate statistics TODO") \ No newline at end of file + if(not args.only_stats): + for session_id in sessions: + + # Declaring initial variables + session = sessions[session_id] + ip = session_ip(session) + timestamp = session_timestamp(session)[:22].replace("T", " ") # clean up the timestamp text all pretty like + durationstring = "closed after " + str(session_duration(session)) + "s" + if(session_duration(session) < 0): + durationstring = "active as of processing" + + if(not session_attempted_login(session)): # Printing empty pings + print(timestamp + " > " + ppip(ip, True, True) + " > Connection opened and " + durationstring) + + elif(session_attempted_login(session) and not session_login_succeeded(session)): # Printing failed logins + print(timestamp + " > " + ppip(ip, True, True) + " > " + ccgen("RED") +"Failed login" + ccgen() + " with credentials: " + str(session_cred_combos(session))) + + elif(session_ran_command(session)): # Printing logic for sessions in which a command was run + + if(not verbosity == 1 and session_is_cmdping(session)): # If -v is not enabled, collapse "command pings" to one-liners + print(timestamp + " > " + ppip(ip, True, True) + " > " + ccgen("GREEN") + "Login" + ccgen() + " with credentials: " + str(session_cred_combos(session)) + ccgen("BLUE") + " ran commands " + ccgen() + str(session_pingcmds(session))) + + else: # Printing logic for un-collapsed sessions where commands were run + print(timestamp + " > " + ppip(ip, True, True) + " > " + ccgen("CYAN") + "Session" + ccgen() + ":") + for entry in session: + match(entry["eventid"]): + case "cowrie.login.failed": + pw = [] + if("password" in entry): + pw = entry["password"] + print(" | Failed login: " + entry["username"] + "/" + str(pw)) + case "cowrie.login.success": + print(" | Successful login: " + entry["username"] + "/" + entry["password"]) + case "cowrie.command.input": + print(" | CMD: " + entry["input"]) + case "cowrie.session.file_download": + df = "" + if("url" in entry): + df = entry["url"] + elif("destfile" in entry): + df = entry["destfile"] + if(verbosity == 1): + print(" | Saved file " + df + " to " + entry["outfile"]) + else: + print(" | Saved file " + df + " to " + entry["outfile"][:40] + "...") + print(" X Session " + durationstring) + + else: # Printing logic for successful logins in which no commands were run + print(timestamp + " > " + ppip(ip, True, True) + " > " + ccgen("GREEN") + "Login" + ccgen() + " with credentials: " + str(session_cred_combos(session))) + if(print_stats): + printstats() -- cgit v1.2.3