KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use add_known_host method in lisa

Best Python code snippet using lisa_python

state.py

Source:state.py

...79                key_pem = arg[8:]80                continue81            if arg.startswith("hostkey-rsa="):82                have_hostkey = True83                self.add_known_host(server, "ssh-rsa", arg[12:])84                continue85            if arg.startswith("hostkey-dss="):86                have_hostkey = True87                self.add_known_host(server, "ssh-dss", arg[12:])88                continue89            if self.use_ecdsa is True:90                if arg.startswith("hostkey-nistp256="):91                    have_hostkey = True92                    self.add_known_host(server, "ecdsa-sha2-nistp256",93                                        arg[17:])94                    continue95                if arg.startswith("hostkey-nistp384="):96                    have_hostkey = True97                    self.add_known_host(server, "ecdsa-sha2-nistp384",98                                        arg[17:])99                    continue100                if arg.startswith("hostkey-nistp521="):101                    have_hostkey = True102                    self.add_known_host(server, "ecdsa-sha2-nistp521",103                                        arg[17:])104                    continue105            else:106                # ECDSA is disabled but the user may have provided hostkeys:107                # just silently ignore such arguments.108                if arg.startswith("hostkey-nistp"):109                    continue110            log.msg("SOCKS: Ignoring invalid argument: " + arg)111        if (user is None or orport is None or key_pem is None or112                not have_hostkey):113            log.msg("SOCKS: Insufficient arguments to create a connection")114            return None115        try:116            self.write_known_hosts()117            self.add_auth_credentials(server, user, key_pem)118        except IOError:119            return None120        # Cache the user/orport so that we can work around #9162121        self.cached_args[server] = {}122        self.cached_args[server]["user"] = user123        self.cached_args[server]["orport"] = orport124        return (user, orport)125    def guess_args(self, server):126        # If we end up here, either the user screwed up the bridge line, or127        # the user's Tor is affected by #9162128        log.msg("SOCKS: No arguments received (Tor bug #9162)")129        if not server in self.known_hosts:130            log.msg("SOCKS: (#9162) Unable to build a known_hosts entry")131            return None132        if not server in self.cached_args:133            log.msg("SOCKS: (#9162) Unable to guess user/OR port")134            return None135        user = self.cached_args[server]["user"]136        key = user + "@" + server137        if not key in self.cached_credentials:138            log.msg("SOCKS: (#9162) No cached RSA key")139            return None140        log.msg("SOCKS: (#9162) Using previously seen values for this host")141        return (user, self.cached_args[server]["orport"])142    def add_known_host(self, host, key_type, key):143        if not key_type in self.key_types:144            if self.use_ecdsa is True:145                if not key_type in self.ecdsa_key_types:146                    log.msg("SOCKS: Invalid ssh host key type: " + key_type)147                    return148            else:149                if key_type in self.ecdsa_key_types:150                    return151                log.msg("SOCKS: Invalid ssh host key type: " + key_type)152                return153        if not host in self.known_hosts:154            self.known_hosts[host] = {}155        if (not key_type in self.known_hosts[host] or156                self.known_hosts[host][key_type] != key):...

test_utils.py

Source:test_utils.py

...69        super().__init__(*args, **kwargs)70        src.utils.KNOWN_HOSTS_PATH = os.environ["KNOWN_HOSTS_PATH"]71        self.KNOWN_HOSTS_PATH = os.environ["KNOWN_HOSTS_PATH"]72    def test_add_host(self):73        add_known_host("github.com")74        with open(self.KNOWN_HOSTS_PATH, "r", encoding="utf-8") as known_hosts:75            lines = known_hosts.readlines()76            self.assertTrue(lines and "ssh-rsa" in lines[0])77    def test_add_host_and_port(self):78        add_known_host("github.com:22")79        with open(self.KNOWN_HOSTS_PATH, "r", encoding="utf-8") as known_hosts:80            lines = known_hosts.readlines()81            self.assertTrue(lines and "ssh-rsa" in lines[0])82    def test_invalid_host(self):83        self.assertRaises(84            InvalidHostError,85            lambda: add_known_host("nonexistinghost123456789123456789.com"),86        )87    def tearDown(self):88        os.remove(self.KNOWN_HOSTS_PATH)89class IsRepoBlacklisted(TestCase):90    blacklist = "facebook/react,nodejs/node,gorilla/websocket"91    def test_blacklisted_repo(self):92        self.assertEqual(is_repo_blacklisted(93            "facebook/react", self.blacklist), True)94    def test_non_blacklisted_repo(self):95        self.assertEqual(is_repo_blacklisted(96            "gatsbyjs/gatsby", self.blacklist), False)97    def test_undefined_blacklist(self):98        self.assertEqual(is_repo_blacklisted("gatsbyjs/gatsby", None), False)99class FixSshPrivateKey(TestCase):...

ssh.py

Source:ssh.py

1#!/usr/bin/env python32"""3SSH Utilities.4"""5import logging6import os7from slurm import expand_nodes, find_all_nodes8from utils import bash, get_argparse9try:10    from tqdm import tqdm11except ImportError:12    def tqdm(itr):13        return itr14logger = logging.getLogger("ssh")15# github's ssh key's. added automatically for convenience.16GITHUB_HOSTS = """17|1|diLJva1ldlvIkh/vsQ4QrfH9D64=|gfYHF6vxRdUd40/dCLlp+z1Ttz8= ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBEmKSENjQEezOmxkZMy7opKgwFB9nkt5YRrYMjNuG5N87uRgg6CLrbo5wAdT/y6v0mKV0U2w0WZ2YB/++Tpockg=18|1|Ha9/dblK4LrKlaqvVlpC9DIVCT4=|dijiEzGXy93Arft/t274K8DswOI= ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBEmKSENjQEezOmxkZMy7opKgwFB9nkt5YRrYMjNuG5N87uRgg6CLrbo5wAdT/y6v0mKV0U2w0WZ2YB/++Tpockg=19"""20def check_for_strict_hosts():21    """22    Ensures the user has set up SSH correctly.23    """24    logger.debug("Checking your ssh config for issues")25    fname = os.path.expanduser("~/.ssh/config")26    with open(fname) as f:27        data = f.read()28        lines = data.split("\n")29    for l1, l2 in zip(lines, lines[1:]):30        if "Host hpc-pg0-*" in l1 and "StrictHostKeyChecking no" in l2:31            logger.debug("Confirmed StrictHostKeyChecking no")32            break33    else:34        logger.warning(35            "Detected you are missing StrictHostKeyChecking. Fixing for you."36        )37        with open(fname, "w") as f:38            f.write(data)39            f.write("\n\n")40            f.write("# ---- set up by azure tools ---\n")41            f.write("Host hpc-pg0-*\n")42            f.write("    StrictHostKeyChecking no\n")43            f.write("# ---- end azure tools ---\n")44def clean_known_hosts():45    """46    Clean up all the known ssh hosts so fingerprints are real nice and up to date.47    """48    check_for_strict_hosts()49    logger.info("Cleaning up your known ssh hosts")50    # put github's keys on the nice list51    with open(os.path.expanduser("~/.ssh/known_hosts"), "w") as f:52        f.write(GITHUB_HOSTS.strip() + "\n")53    # add everyone's keys to the nice list54    logger.debug("Adding everyone's ssh fingerpint to your list")55    add_known_host = os.path.join(os.path.dirname(__file__), "add_known_host.sh")56    for host in expand_nodes(find_all_nodes()):57        try:58            bash(f"{add_known_host} {host}", silent_stderr=True)59        except:60            logging.error(f"Couldn't add host {host} to known hosts.")61def _pdsh_unsafe(cmd: str, hosts: str, silent_stderr: bool = False):62    """63    Runs the command on all hosts, and returns the output, ala pdsh.64    Hosts should be in '-w' format, e.g. 'hpc-pg0-[1,3-5]'.65    This version differs from the public "pdsh" version in that it66    does not perform ssh checks.67    """68    TIMEOUT = 6069    logger.debug(f"Running global pdsh (hosts: {hosts}) with cmd {cmd}")70    # note -u timeout flag only marks the cmd as a failure on the launcher side.71    # the command may continue to run indefinitely on the worker. see72    # `man pdsh` LIMITATIONS section.73    # however, we DO need the timeout on this side so we don't hang just because74    # one worker is having a bad day.75    return bash(76        f"pdsh -R ssh -u {TIMEOUT} -w '{hosts}' {cmd}", silent_stderr=silent_stderr77    )78def pdsh(cmd: str, hosts: str, silent_stderr: bool = False):79    check_for_strict_hosts()80    return _pdsh_unsafe(cmd, hosts, silent_stderr=silent_stderr)81def main():82    argparse = get_argparse()83    _args = argparse.parse_args()  # for verbose mode84    clean_known_hosts()85if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.