Skip to content

Instantly share code, notes, and snippets.

Last active May 29, 2020 08:46
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save athoune/8190f92fe954563cf520225319019708 to your computer and use it in GitHub Desktop.
nia nia nia ssh public key
#!/usr/bin/env python3
import pwd
import sys
import re
import subprocess
import tempfile
import os
import io
from collections import namedtuple
import socket
ACCEPTED = re.compile(r"Accepted publickey for ([^ ]+) from ([^ ]+) port (\d+) ssh2: (\w+) ([^ ]+)")
OPTIONS = re.compile(r'^((([a-z\-]+="[^"]+")|([^ ]+))+) ')
Key = namedtuple('Key', ['type', 'key', 'options', 'comment'])
Fingerprint = namedtuple('Fingerprint', ['size', 'hash', 'type'])
def ssh_key(line):
KEYS = ['', 'ecdsa-sha2-nistp256', 'ecdsa-sha2-nistp384',
'ecdsa-sha2-nistp521', '', 'ssh-ed25519', 'ssh-dss', 'ssh-rsa']
options = None
if not line.split(' ', 1)[0] in KEYS:
m = OPTIONS.match(line)
if m:
options =
line = line[len(options):]
slugs = line.strip().split(' ', 2)
assert slugs[0] in KEYS, slugs[0]
comment = None
if len(slugs) == 3:
comment = slugs[2]
return Key(slugs[0], slugs[1], options, comment)
def keys(user):
u = pwd.getpwnam(user)
with open(u.pw_dir + '/.ssh/authorized_keys', 'r') as f:
for line in f.readlines():
if line.strip() == '' or line.startswith('#'):
k = ssh_key(line)
fd, tmp = tempfile.mkstemp()
f = open(fd, 'w')
with subprocess.Popen(['ssh-keygen', '-l', '-f', tmp], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as proc:
rc = proc.wait()
if rc != 0:
raise Exception(line,
r =
rr = r.strip().decode('utf8').split(' ')
yield Fingerprint(int(rr[0]), rr[1], rr[-1][1:-1]), k.comment
class Keys:
def __init__(self):
self._users = dict()
def __call__(self, user, fingerprint):
if user not in self._users:
self._users[user] = dict()
for fp, comment in keys(user):
self._users[user][fp.hash] = comment
return self._users[user][fingerprint]
class Abuse:
def __init__(self):
self._abuses = dict()
def __call__(self, ip):
if ip not in self._abuses:
with subprocess.Popen(['whois', '-b', ip], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as proc:
rc = proc.wait()
if rc != 0:
raise Exception(ip,
for line in proc.stdout:
line = line.decode('utf8')
if line.startswith("abuse-mailbox"):
self._abuses[ip] = re.compile(r"\s+").split(line[:-1])[-1]
return self._abuses.get(ip)
def read_auth_log():
keys = Keys()
abuse = Abuse()
for line in sys.stdin.readlines():
slugs = line[:-1].split(" ", 5)
if not slugs[4].startswith("sshd"):
comment = slugs[5]
if not comment.startswith("Accepted publickey"):
m = ACCEPTED.match(comment)
user =
fp =
name = None
name, alias, addresslist = socket.gethostbyaddr(
except Exception as e:
#print(, e)
name = abuse(
print(keys(user, fp), name)
except Exception as e:
print("aaaaaaah", e)
if __name__ == "__main__":
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment