<commit_before>#!/usr/bin/python3
import sys, os, pty
sys.path.append(os.path.join(os.path.dirname(__file__),'../bindings/python/'))
import serverboards, pexpect, shlex, re, subprocess, random
import urllib.parse as urlparse
from common import *
from serverboards import print

@serverboards.rpc_method
def ssh_public_key(**kwargs):
    ensure_ID_RSA()
    with open(ID_RSA_PUB, 'r') as fd:
        return fd.read()

def ssh_urlparse(url):
    if not "://" in url:
        url="ssh://%s"%(url)
    u = urlparse.urlparse(url)
    assert u.scheme == 'ssh'
    port = u.port or '22'
    return (u.hostname, str(port))


def get_fingerprint(url, *args, **kwargs):
    if not url:
        return None
    (hostname, port) = ssh_urlparse(url)
    try:
        output = str(subprocess.check_output(["ssh-keyscan", "-p", port, hostname]), 'utf8')
        output=output.strip().split('\n')
        output.sort()
        # serverboards.debug(repr(output))
        return '\n'.join(output)
    except:
        import traceback; traceback.print_exc()
        return None

@serverboards.rpc_method
def remote_fingerprint(url="", options="", **kwargs):
    # serverboards.info(repr(kwargs))
    fingerprint=get_fingerprint(url, options)
    if not fingerprint:
        return {
            "fingerprint": "Cant connect to server at <%s>. Set a valid SSH address."%url,
            "className": "grey disabled",
            "toggle": "Select a valid SSH address",
            "enabled": False
        }
    fingerprintb = bytes(fingerprint, 'utf8')

    sp=subprocess.Popen(["ssh-keygen", "-lf", "/dev/stdin", "-E", "md5"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    fingerprint_md5, error = sp.communicate(fingerprintb)

    sp=subprocess.Popen(["ssh-keygen", "-lf", "/dev/stdin", "-E", "sha1"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    fingerprint_sha1, error = sp.communicate(fingerprintb)

    sp=subprocess.Popen(["ssh-keygen", "-lf", "/dev/stdin", "-E", "sha256"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    fingerprint_sha256, error = sp.communicate(fingerprintb)

    fingerprint_text = ''.join( [str(fingerprint_md5, 'utf8'), "\n", str(fingerprint_sha1, 'utf8'), "\n", str(fingerprint_sha256, 'utf8')] )

    enabled=False
    if os.path.exists(KNOWN_HOSTS_FILE):
        with open(KNOWN_HOSTS_FILE,"r") as fd:
            enabled = fingerprint in fd.read()

    return {
        "fingerprint_orig" : fingerprint,
        "fingerprint": fingerprint_text,
        "toggle" : "Disable host" if enabled else "Enable host",<commit_msg>Added timeout to check fingerprint<commit_after>#!/usr/bin/python3
import sys, os, pty
sys.path.append(os.path.join(os.path.dirname(__file__),'../bindings/python/'))
import serverboards, pexpect, shlex, re, subprocess, random
import urllib.parse as urlparse
from common import *
from serverboards import print

@serverboards.rpc_method
def ssh_public_key(**kwargs):
    ensure_ID_RSA()
    with open(ID_RSA_PUB, 'r') as fd:
        return fd.read()

def ssh_urlparse(url):
    if not "://" in url:
        url="ssh://%s"%(url)
    u = urlparse.urlparse(url)
    assert u.scheme == 'ssh'
    port = u.port or '22'
    return (u.hostname, str(port))


def get_fingerprint(url, *args, **kwargs):
    if not url:
        return None
    (hostname, port) = ssh_urlparse(url)
    try:
        output = str(subprocess.check_output(["ssh-keyscan", "-p", port, hostname], timeout=5), 'utf8')
        output=output.strip().split('\n')
        output.sort()
        # serverboards.debug(repr(output))
        return '\n'.join(output)
    except subprocess.TimeoutExpired:
        return None
    except:
        import traceback; traceback.print_exc()
        return None

@serverboards.rpc_method
def remote_fingerprint(url="", options="", **kwargs):
    # print(url, options, repr(kwargs))
    fingerprint=get_fingerprint(url, options)
    if not fingerprint:
        return {
            "fingerprint": "Cant connect to server at <%s>. Set a valid SSH address."%url,
            "className": "grey disabled",
            "toggle": "Select a valid SSH address",
            "enabled": False
        }
    fingerprintb = bytes(fingerprint, 'utf8')

    sp=subprocess.Popen(["ssh-keygen", "-lf", "/dev/stdin", "-E", "md5"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    fingerprint_md5, error = sp.communicate(fingerprintb)

    sp=subprocess.Popen(["ssh-keygen", "-lf", "/dev/stdin", "-E", "sha1"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    fingerprint_sha1, error = sp.communicate(fingerprintb)

    sp=subprocess.Popen(["ssh-keygen", "-lf", "/dev/stdin", "-E", "sha256"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    fingerprint_sha256, error = sp.communicate(fingerprintb)

    fingerprint_text = ''.join( [str(fingerprint_md5, 'utf8'), "\n", str(fingerprint_sha1, 'utf8'), "\n", str(fingerprint_sha256, 'utf8')] )

    enabled=False
    if os.path.exists(KNOWN_HOSTS_FILE):
        with open(KNOWN_HOSTS_FILE,"r") as fd:
            enabled = fingerprint in fd.read()

    return {
        "fingerprint_orig" : fingerprint,
        "fingerprint": fingerprint_text,
        "toggle" : "Disable host" if enabled else "Enable host",
{'fraction_non_alphanumeric': 0.15267616480554486, 'fraction_numerical': 0.009048902579899884, 'mean_word_length': 3.402542372881356, 'pattern_counts': {'":': 12, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1202282', 'n_tokens_mistral': 1819, 'n_tokens_neox': 1720, 'n_words': 382}