Python script to check if a system is vulnerable to SSHole
#!/usr/bin/env python3
"""Audit a host, its containers or a remote host for the sshole attack on OpenSSH (CVE-2026-123158).
The attack needs all of: a musl-linked sshd, UsePAM off, PasswordAuthentication on, SHA-256/512
password hashes in /etc/shadow, and a no-password ('*') account sshd lets log in. A 257-1024
character password makes musl's crypt() fail and return '*', which matches that account.
sshole-check.py this host
sshole-check.py --docker|--podman every running container on this host
sshole-check.py --ssh [user@]host a remote host (add --docker/--podman for its containers)
Containers are inspected from the host (chroot into /proc/<pid>/root), so images
without sh or any shell tools are covered. Needs root for containers and for the
sshd -T and /etc/shadow checks. Remote runs send this script over ssh stdin; the
remote needs python3 (3.6+).
Exit status: 0 not exploitable, 1 VULNERABLE or POSSIBLE (a condition couldn't be checked),
2 a target couldn't be inspected.
"""
import argparse
import fnmatch
import glob
import json
import os
import struct
import subprocess
import sys
import time
SSHD_PATHS = ('/usr/sbin/sshd', '/usr/bin/sshd', '/sbin/sshd', '/usr/local/sbin/sshd')
SSHD_COMMS = ('sshd', 'sshd-session', 'sshd-auth')
APK_DBS = ('/lib/apk/db/installed', '/usr/lib/apk/db/installed')
APK_PKGS = ('openssh-server', 'openssh-server-pam')
RPM_DIRS = ('/var/lib/rpm', '/usr/lib/sysimage/rpm')
RPM_QF = r'%{NAME}-%{VERSION}-%{RELEASE}.%{ARCH}\n'
def elf_interp(path):
"""The ELF interpreter (dynamic loader) of a binary, '' if static, None if not ELF."""
with open(path, 'rb') as f:
hdr = f.read(64)
if hdr[:4] != b'\x7fELF':
return None
end = '<' if hdr[5] == 1 else '>'
is64 = hdr[4] == 2
if is64:
phoff = struct.unpack(end + 'Q', hdr[32:40])[0]
phentsize, phnum = struct.unpack(end + 'HH', hdr[54:58])
else:
phoff = struct.unpack(end + 'I', hdr[28:32])[0]
phentsize, phnum = struct.unpack(end + 'HH', hdr[42:46])
for i in range(phnum):
f.seek(phoff + i * phentsize)
ph = f.read(phentsize)
if struct.unpack(end + 'I', ph[:4])[0] != 3: # PT_INTERP
continue
if is64:
off, size = struct.unpack(end + 'Q', ph[8:16])[0], struct.unpack(end + 'Q', ph[32:40])[0]
else:
off, size = struct.unpack(end + 'I', ph[4:8])[0], struct.unpack(end + 'I', ph[16:20])[0]
f.seek(off)
return f.read(size).rstrip(b'\0').decode('ascii', 'replace')
return ''
def read_text(path):
try:
with open(path, errors='replace') as f:
return f.read()
except OSError:
return None
def distro_name():
alpine = read_text('/etc/alpine-release')
if alpine:
return 'Alpine ' + alpine.strip()
osr = read_text('/etc/os-release') or read_text('/usr/lib/os-release') or ''
fields = dict(line.split('=', 1) for line in osr.splitlines() if '=' in line)
name = fields.get('ID', '').strip('"')
ver = fields.get('VERSION_ID', '').strip('"')
return (name + ' ' + ver).strip() or 'unknown'
def libc_from_loaders():
if glob.glob('/lib/ld-musl-*.so.1'):
return 'musl'
if glob.glob('/lib*/ld-linux*.so.*') or glob.glob('/lib/*/ld-linux*.so.*'):
return 'glibc'
return 'unknown'
def rpm_header_tags(blob, wanted):
"""String tags from an rpm header blob (no lead/magic, as stored in rpmdb.sqlite)."""
il, dl = struct.unpack('>II', blob[:8])
data = blob[8 + 16 * il:8 + 16 * il + dl]
out = {}
for i in range(il):
tag, typ, off, _ = struct.unpack('>IIII', blob[8 + 16 * i:24 + 16 * i])
if tag in wanted and typ == 6: # RPM_STRING_TYPE
out[tag] = data[off:data.index(b'\0', off)].decode('utf-8', 'replace')
return out
def rpm_packages():
"""(packages, note): via the rpm binary if present, else by reading rpmdb.sqlite directly."""
db_dir = next((d for d in RPM_DIRS if os.path.isdir(d) and os.listdir(d)), None)
if not db_dir:
return [], None
for rpm in ('/usr/bin/rpm', '/bin/rpm'):
if not os.path.isfile(rpm):
continue
try:
p = subprocess.run([rpm, '-q', '--qf', RPM_QF, 'openssh-server'],
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, universal_newlines=True, timeout=60)
except (OSError, subprocess.TimeoutExpired) as e:
return [], 'rpm -q failed: %s' % e
# rc 1 with 'is not installed' on stdout means absent; anything else is a real failure
if p.returncode == 0:
return p.stdout.split(), None
if 'not installed' in p.stdout:
return [], None
return [], 'rpm -q failed: %s' % (p.stderr.strip().splitlines() or ['exit %d' % p.returncode])[0]
sqlite_db = os.path.join(db_dir, 'rpmdb.sqlite')
if not os.path.isfile(sqlite_db):
return [], 'rpm database in %s but no rpm binary and no rpmdb.sqlite (BDB/ndb not readable)' % db_dir
try:
import sqlite3
con = sqlite3.connect('file:%s?mode=ro&immutable=1' % sqlite_db, uri=True)
try:
blobs = [r[0] for r in con.execute('SELECT blob FROM Packages')]
finally:
con.close()
found = []
for blob in blobs:
t = rpm_header_tags(bytes(blob), (1000, 1001, 1002, 1022)) # NAME VERSION RELEASE ARCH
if t.get(1000) == 'openssh-server':
found.append('%s-%s-%s.%s' % (t[1000], t.get(1001, '?'), t.get(1002, '?'), t.get(1022, '?')))
return found, None
except Exception as e:
return [], 'reading %s failed: %s: %s' % (sqlite_db, type(e).__name__, e)
def packages():
"""(openssh-server packages, note) from the apk, dpkg or rpm database."""
found = []
for db in APK_DBS:
text = read_text(db)
if not text:
continue
for block in text.split('\n\n'):
f = dict(l.split(':', 1) for l in block.splitlines() if l[1:2] == ':')
if f.get('P') in APK_PKGS:
found.append('%s-%s' % (f['P'], f.get('V', '?')))
break
text = read_text('/var/lib/dpkg/status')
if text:
for block in text.split('\n\n'):
f = dict(l.split(': ', 1) for l in block.splitlines() if ': ' in l and not l.startswith(' '))
if f.get('Package') == 'openssh-server' and 'installed' in f.get('Status', '').split():
found.append('openssh-server_%s' % f.get('Version', '?'))
rpms, note = rpm_packages()
return found + rpms, note
def sshd_effective(sshd):
"""(settings, error): sshd -T as {key: [values]}, values of repeated keys collected."""
try:
p = subprocess.run([sshd, '-T'], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, universal_newlines=True, timeout=15)
except (OSError, subprocess.TimeoutExpired) as e:
return None, 'sshd -T failed: %s' % e
if p.returncode != 0:
return None, 'sshd -T failed: %s' % (p.stderr.strip().splitlines() or ['exit %d' % p.returncode])[0]
cfg = {}
for line in p.stdout.splitlines():
key, _, val = line.partition(' ')
cfg.setdefault(key.lower(), []).extend(val.split())
return cfg, None
def match_password_auth():
"""True if any Match block in sshd_config (or its Includes) turns PasswordAuthentication on.
sshd -T shows only the global settings; Match can't change UsePAM, but it can enable passwords.
"""
found, seen = False, set()
def scan(path):
nonlocal found
if path in seen:
return
seen.add(path)
in_match = False
for raw in (read_text(path) or '').splitlines():
words = raw.split('#', 1)[0].split()
if not words:
continue
key = words[0].lower()
if key == 'include':
for pat in words[1:]:
for inc in sorted(glob.glob(pat if pat.startswith('/') else '/etc/ssh/' + pat)):
scan(inc)
elif key == 'match':
in_match = True
elif in_match and key == 'passwordauthentication' and words[1:2] == ['yes']:
found = True
scan('/etc/ssh/sshd_config')
return found
def read_shadow():
"""[(user, hash, expire_day or None)], or None if unreadable."""
text = read_text('/etc/shadow')
if text is None:
return None
out = []
for line in text.splitlines():
p = line.split(':')
if len(p) >= 2:
exp = p[7] if len(p) > 7 else ''
out.append((p[0], p[1], int(exp) if exp.isdigit() else None))
return out
def read_passwd_groups():
"""({user: [primary gid, shell]}, {gid: [group name, [members]]})."""
users, groups = {}, {}
for line in (read_text('/etc/passwd') or '').splitlines():
p = line.split(':')
if len(p) >= 7:
users[p[0]] = [p[3], p[6]]
for line in (read_text('/etc/group') or '').splitlines():
p = line.split(':')
if len(p) >= 4:
groups[p[2]] = [p[0], [m for m in p[3].split(',') if m]]
return users, groups
def proc_sshd_pids():
"""sshd PIDs from this root's /proc: inside a container root that's the container's own procfs."""
if not os.path.isdir('/proc/1'): # no procfs here (our own /proc/self isn't in a container's)
return None
pids = []
for d in os.listdir('/proc'):
try:
with open('/proc/%s/comm' % d) as f:
if d.isdigit() and f.read().strip() in SSHD_COMMS:
pids.append(int(d))
except OSError:
continue
return pids
def collect(in_container=False):
"""Everything checked from the filesystem, run with / as the target's root."""
os.environ['PATH'] = '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'
pkgs, pkg_note = packages()
users, groups = read_passwd_groups()
info = {'distro': distro_name(), 'libc': libc_from_loaders(), 'packages': pkgs, 'pkg_note': pkg_note,
'sshd': None, 'interp': None, 'config': None, 'config_error': None,
'match_pwauth': match_password_auth(), 'shadow': read_shadow(), 'users': users,
'groups': groups, 'root': os.geteuid() == 0}
for path in SSHD_PATHS:
if os.path.isfile(path):
info['sshd'] = path
try:
info['interp'] = elf_interp(path)
except OSError as e:
info['interp'] = 'unreadable: %s' % e
info['config'], info['config_error'] = sshd_effective(path)
break
if in_container:
info['pids'] = proc_sshd_pids()
return info
def collect_chrooted(root):
"""Run collect() in a child chrooted into a container's root, so symlinks resolve inside it."""
r, w = os.pipe()
pid = os.fork()
if pid == 0:
os.close(r)
try:
os.chroot(root)
os.chdir('/')
data = collect(in_container=True)
except Exception as e:
data = {'error': '%s: %s' % (type(e).__name__, e)}
with os.fdopen(w, 'w') as f:
json.dump(data, f)
os._exit(0)
os.close(w)
with os.fdopen(r) as f:
raw = f.read()
os.waitpid(pid, 0)
try:
return json.loads(raw)
except ValueError:
return {'error': 'no result from the chrooted child'}
def sshd_pids(ns_pid):
"""PIDs of sshd processes in the given PID namespace (via the host's /proc)."""
try:
target = os.readlink('/proc/%d/ns/pid' % ns_pid)
except OSError:
return None
pids, denied = [], False
for d in os.listdir('/proc'):
if not d.isdigit():
continue
try:
if os.readlink('/proc/%s/ns/pid' % d) != target:
continue
with open('/proc/%s/comm' % d) as f:
if f.read().strip() in SSHD_COMMS:
pids.append(int(d))
except PermissionError:
denied = True
except OSError:
continue
# Without root, other users' processes can't be placed in a namespace: don't claim 'No'.
return None if denied and not pids else pids
NO_SHELLS = ('nologin', 'false', 'true', 'sync', 'shutdown', 'halt')
def sshd_libc(info):
interp = info.get('interp') or ''
if 'ld-musl' in interp:
return 'musl'
if 'ld-linux' in interp:
return 'glibc'
return info['libc'] # static, unreadable or no sshd binary: fall back to the system loader
def user_allowed(user, cfg, users, groups):
"""sshd's DenyUsers/AllowUsers/DenyGroups/AllowGroups for a user. user@host patterns count as
a match for Allow and as no match for Deny: the source host is unknown, so assume the worst."""
def match(pats, name, deny):
for pat in pats:
name_pat, at, _ = pat.partition('@')
if fnmatch.fnmatchcase(name, name_pat) and not (deny and at):
return True
return False
gid, _ = users.get(user, [None, ''])
member_of = [g[0] for k, g in groups.items() if k == gid or user in g[1]]
if match(cfg.get('denyusers', []), user, True):
return False
if cfg.get('allowusers') and not match(cfg['allowusers'], user, False):
return False
if any(match(cfg.get('denygroups', []), g, True) for g in member_of):
return False
if cfg.get('allowgroups') and not any(match(cfg['allowgroups'], g, False) for g in member_of):
return False
return True
def assess(info, today=None):
"""Weigh the CVE-2026-123158 conditions. Returns a dict for report().
All must hold: musl sshd, PAM off, PasswordAuthentication on, SHA-256/512 hashes present,
and a '*' (no password) account sshd will let log in. Each condition is True, False or None (unknown).
"""
if today is None:
today = int(time.time() // 86400)
libc = sshd_libc(info)
res = {'libc': libc, 'verdict': None, 'conds': [], 'targets': [], 'noshell': []}
if not info['sshd'] and not info['packages']:
res['verdict'] = 'no sshd'
return res
cfg = info['config']
conds = res['conds']
conds.append(('musl sshd (%s)' % libc, {'musl': True, 'glibc': False}.get(libc)))
if cfg is None:
conds.append(('UsePAM no', None))
conds.append(('PasswordAuthentication yes', None))
else:
# no usepam line at all means sshd was built without PAM (Alpine's plain openssh-server)
usepam = (cfg.get('usepam') or ['no (not built in)'])[0]
conds.append(('UsePAM no [%s]' % usepam, usepam != 'yes'))
pw = (cfg.get('passwordauthentication') or ['yes'])[0]
if pw == 'yes':
conds.append(('PasswordAuthentication yes', True))
elif info['match_pwauth']:
conds.append(('PasswordAuthentication yes [no globally, yes in a Match block]', None))
else:
conds.append(('PasswordAuthentication yes [%s]' % pw, False))
shadow = info['shadow']
if shadow is None:
why = '/etc/shadow not readable' + ('' if info['root'] else ', not root')
conds.append(('SHA-256/512 hashes present [%s]' % why, None))
conds.append(('loginable no-password (*) accounts [%s]' % why, None))
else:
hashes = sum(1 for _, h, _ in shadow if h.startswith(('$5$', '$6$')))
conds.append(('SHA-256/512 hashes present [%d]' % hashes, hashes > 0))
root_ok = cfg is None or (cfg.get('permitrootlogin') or ['yes'])[0] == 'yes'
for user, h, exp in shadow:
if h != '*' or (exp is not None and exp <= today):
continue
if user == 'root' and not root_ok:
continue
if cfg is not None and not user_allowed(user, cfg, info['users'], info['groups']):
continue
shell = info['users'].get(user, ['', '/bin/sh'])[1] or '/bin/sh'
(res['noshell'] if os.path.basename(shell) in NO_SHELLS else res['targets']).append(user)
conds.append(('loginable no-password (*) accounts with a shell [%d]' % len(res['targets']),
bool(res['targets'])))
states = [ok for _, ok in conds]
if False in states:
res['verdict'] = 'not exploitable' if libc == 'musl' else 'ok (%s)' % libc
elif None in states:
res['verdict'] = 'POSSIBLE'
else:
res['verdict'] = 'VULNERABLE'
return res
ROW = '%-28s %-14s %-6s %-44s %-8s %s'
MARK = {True: '[yes]', False: '[ no]', None: '[ ? ]'}
def report(name, info, pids):
"""Print one target; returns 0 clean, 1 vulnerable or possibly, 2 couldn't inspect."""
if 'error' in info:
print(ROW % (name, '-', '-', '-', '-', 'UNABLE: ' + info['error']))
return 2
a = assess(info)
running = '?' if pids is None else ('Running' if pids else 'No')
verdict = a['verdict']
if verdict in ('VULNERABLE', 'POSSIBLE') and pids == []:
verdict += ' (sshd not running)'
print(ROW % (name[:28], info['distro'][:14], a['libc'], ','.join(info['packages']) or 'None',
running, verdict))
details = []
if info.get('pkg_note'):
details.append('packages: ' + info['pkg_note'])
if info['sshd']:
interp = info['interp']
details.append('sshd: %s (%s)' % (info['sshd'], 'static' if interp == '' else 'loader ' + str(interp)))
if info.get('config_error'):
details.append(info['config_error'])
if pids:
details.append('sshd pids: ' + ' '.join(map(str, pids)))
for label, ok in a['conds']:
details.append('%s %s' % (MARK[ok], label))
exposed = a['libc'] != 'glibc' # on glibc, crypt() doesn't fail this way: the lists would mislead
if exposed and a['targets']:
details.append(' no-password accounts sshd would accept: ' + ', '.join(a['targets']))
if exposed and a['noshell']:
details.append(' also accepted, but with no login shell (forwarding only): ' + ', '.join(a['noshell']))
for d in details:
print(' ' + d)
return 1 if a['verdict'] in ('VULNERABLE', 'POSSIBLE') else 0
def header():
print(ROW % ('TARGET', 'DISTRO', 'LIBC', 'OPENSSH PACKAGE', 'SSHD', 'VERDICT'))
print('-' * 120)
def audit_local():
header()
return report(os.uname()[1] + ' (host)', collect(), sshd_pids(os.getpid()))
def audit_containers(runtime):
try:
ids = subprocess.run([runtime, 'ps', '-q'], stdout=subprocess.PIPE, check=True,
universal_newlines=True).stdout.split()
except (OSError, subprocess.CalledProcessError) as e:
print('ERROR: %s ps failed: %s' % (runtime, e), file=sys.stderr)
return 2
header()
if not ids:
print('(no running containers)')
return 0
out = subprocess.run([runtime, 'inspect', '-f', '{{.Name}}|{{.State.Pid}}'] + ids,
stdout=subprocess.PIPE, universal_newlines=True).stdout.splitlines()
worst = 0
for line in out:
name, _, pid = line.rpartition('|')
name = name.lstrip('/')
if not pid.isdigit() or pid == '0':
worst = max(worst, report(name, {'error': 'not running'}, None))
continue
pid = int(pid)
info = collect_chrooted('/proc/%d/root' % pid)
if 'error' in info and os.geteuid() != 0:
info['error'] += ' (run as root; for rootless podman: podman unshare %s --podman)' % sys.argv[0]
worst = max(worst, report(name, info, info.get('pids')))
return worst
def audit_ssh(target, mode, sudo):
try:
with open(os.path.abspath(sys.argv[0]), 'rb') as f:
src = f.read()
except OSError as e:
print('ERROR: cannot read this script to send it: %s' % e, file=sys.stderr)
return 2
remote = ('sudo -n ' if sudo else '') + 'python3 - ' + (mode or '')
p = subprocess.run(['ssh', '-o', 'BatchMode=yes', target, remote], input=src)
if p.returncode == 127:
print('ERROR: python3 not found on %s' % target, file=sys.stderr)
return 2
if p.returncode == 255:
print('ERROR: ssh to %s failed' % target, file=sys.stderr)
return 2
return p.returncode
def main():
ap = argparse.ArgumentParser(description=__doc__.split('\n')[0])
rt = ap.add_mutually_exclusive_group()
rt.add_argument('--docker', action='store_true', help='audit running docker containers')
rt.add_argument('--podman', action='store_true', help='audit running podman containers')
ap.add_argument('--ssh', metavar='[USER@]HOST', help='run the audit on a remote host')
ap.add_argument('--sudo', action='store_true', help='with --ssh: run as root via passwordless sudo')
args = ap.parse_args()
mode = '--docker' if args.docker else '--podman' if args.podman else None
if args.ssh:
return audit_ssh(args.ssh, mode, args.sudo)
if mode:
return audit_containers(mode[2:])
return audit_local()
if __name__ == '__main__':
sys.exit(main())