ssh_checker.py

import os.path

import badkeys
import csv
from badkeys import allchecks
from tqdm import tqdm
from gitlab_connector import get_all_users, get_keys_for_user

class SSHKey:
    def __init__(self, key_id, key_name, algo):
        self.key_id = key_id
        self.key_name = key_name
        self.algo = algo
        self.failed_checks = {}

    def add_failed_check(self, check_name, subtest=None, desc=None):
        self.failed_checks[check_name] = {"subtest": subtest, "desc": desc}

    def has_issues(self):
        return bool(self.failed_checks)


class User:
    def __init__(self, user_id, name):
        self.user_id = user_id
        self.name = name
        self.keys = []

    def add_key(self, ssh_key):
        self.keys.append(ssh_key)

    def get_insecure_keys(self):
        return [key for key in self.keys if key.has_issues()]

# Fetch all users from Gitlab
print("Fetching all users from Gitlab...")
users = get_all_users()
algorithm_usage = {}

print(f"Found {len(users)} users.")
print("Due to rate limiting, the Gitlab server might not allow to fetch all user keys within one run of this script.")
start_index = int(input("From which index in the user list do you want to start?\n"))
end_index = int(input("At which index in the user list do you want to stop?\n"))

users = users[start_index:end_index]

print("Output will be written to ssh-check-results.csv")

# If output file doesn't exist yet, create it and write header line
if not os.path.exists("ssh-check-results.csv"):
    with open('ssh-check-results.csv', 'w', newline='') as result_file:
        csv.writer(result_file, delimiter=";").writerow(['user-name', 'user-id', 'key-name', 'key-id', 'algorithm',
                                                        'check', 'subcheck', 'description'])

for user_data in tqdm(users, desc="Checking keys of users"):
    user = User(user_data.id, user_data.name)

    # Get all ssh keys of current user
    keys = get_keys_for_user(user.user_id)
    for key_data in keys:
        # Run badkeys on current key
        check_result = badkeys.checksshpubkey(key_data.key, allchecks)

        # Get and store algorithm of current key
        if check_result["type"] == "rsa":
            algo = "rsa" + str(check_result.get("bits", ""))
        elif check_result["type"] == "ec" and "y" not in check_result:
            algo = "Ed25519"
        else:
            algo = check_result["type"]

        algorithm_usage[algo] = algorithm_usage.get(algo, 0) + 1

        ssh_key = SSHKey(key_data.id, key_data.title, algo)


        # Add failed checks to the key
        for check_name, details in check_result["results"].items():

            # If something is wrong with the RSA exponent, I want to know which exponent it is
            if "exponent" in details.get("subtest"):
                e = check_result["e"]
                details["desc"] = f"e={e}" + details.get("desc", "")

            ssh_key.add_failed_check(check_name, details.get("subtest"), details.get("desc"))

        user.add_key(ssh_key)

    insecure_keys = user.get_insecure_keys()

    # Output
    if insecure_keys:
        print(f"\n\033[31mInsecure keys for user {user.name} (id={user.user_id}): {len(insecure_keys)}\033[0m")
        for key in insecure_keys:
            # Build string for failed checks: checkName-subcheck (desc), ...
            failed_checks_string = ""
            first = True
            for check_name, details in key.failed_checks.items():
                if first:
                    failed_checks_string += f"{check_name}"
                    first = False
                else:
                    failed_checks_string += f", {check_name}"
                if details["subtest"] is not None:
                    failed_checks_string += f"-{details["subtest"]}"
                if details["desc"] is not None:
                    failed_checks_string += f" ({details["desc"]})"

                with open('ssh-check-results.csv', 'a', newline='') as result_file:
                    csv.writer(result_file, delimiter=";").writerow([
                        user.name,
                        user.user_id,
                        key.key_name,
                        key.key_id,
                        key.algo,
                        check_name,
                        details.get("subtest", ""),
                        details.get("desc", "")
                    ])

            print(f"\t{key.key_name} (id={key.key_id}) ({key.algo}): {failed_checks_string}")

# Output algorithm distribution:
print("\nAlgorithm distribution:")
for algo, amount in algorithm_usage.items():
    print(f"{algo}: {amount}")