我们从Python开源项目中，提取了以下22个代码示例，用于说明如何使用libvirt.VIR_CRED_AUTHNAME。
def __enter__(self): try: if self.sasl_username and self.sasl_password: def request_cred(credentials, user_data): for credential in credentials: if credential[0] == libvirt.VIR_CRED_AUTHNAME: credential[4] = self.sasl_username elif credential[0] == libvirt.VIR_CRED_PASSPHRASE: credential[4] = self.sasl_password return 0 auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE], request_cred, None] flags = libvirt.VIR_CONNECT_RO if self.readonly else 0 self.conn = libvirt.openAuth(self.uri, auth, flags) elif self.readonly: self.conn = libvirt.openReadOnly(self.uri) else: self.conn = libvirt.open(self.uri) return self.conn except libvirt.libvirtError as e: raise exception.LibvirtConnectionOpenError(uri=self.uri, error=e)
def __init__(self, uri, module): self.module = module cmd = "uname -r" rc, stdout, stderr = self.module.run_command(cmd) if "xen" in stdout: conn = libvirt.open(None) elif "esx" in uri: auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_NOECHOPROMPT], [], None] conn = libvirt.openAuth(uri, auth) else: conn = libvirt.open(uri) if not conn: raise Exception("hypervisor connection failure") self.conn = conn
def get_connection(self): """ Get connection with libvirt using QEMU driver and system context :return conn: connection with libvirt :rtype conn: libvirt connection """ creds = self.args.credentials if 'username' in creds: auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_NOECHOPROMPT], creds['username'], None] conn = libvirt.openAuth(creds['qemu_instance'], auth, 0) else: conn = libvirt.open(creds['qemu_instance']) if conn == None: LOG.error('Failed to open connection to %s', creds['qemu_instance']) LOG.warn('check PAWS documentation %s' % LIBVIRT_AUTH_HELP) raise PawsPreTaskError LOG.debug("connected successfully to %s" % creds['qemu_instance']) return conn
def retrieve_credentials(self, credentials, user_data): """ Retrieves the libvirt credentials in a strange format and hand it to the API in order to communicate with the hypervisor. To be honest, I have no idea why this has to be done this way. I have taken this function from the official libvirt documentation. :param credentials: libvirt credentials object :param user_data: some data that will never be used :type user_data: None """ #get credentials for libvirt for credential in credentials: if credential[0] == libvirt.VIR_CRED_AUTHNAME: credential[4] = self.USERNAME if len(credential[4]) == 0: credential[4] = credential[3] elif credential[0] == libvirt.VIR_CRED_PASSPHRASE: credential[4] = self.PASSWORD else: return -1 return 0
def __libvirt_auth_credentials_callback(self, credentials, user_data): for credential in credentials: if credential[0] == libvirt.VIR_CRED_AUTHNAME: credential[4] = self.login if len(credential[4]) == 0: credential[4] = credential[3] elif credential[0] == libvirt.VIR_CRED_PASSPHRASE: credential[4] = self.passwd else: return -1 return 0
def __connect_tcp(self): flags = [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE] auth = [flags, self.__libvirt_auth_credentials_callback, None] uri = 'qemu+tcp://%s/system' % self.host try: return libvirt.openAuth(uri, auth, 0) except libvirtError as e: self.last_error = 'Connection Failed: ' + str(e) self.connection = None
def __connect_tcp(self): flags = [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE] auth = [flags, self.__libvirt_auth_credentials_callback, None] uri = 'qemu+tcp://%s/system' % self.host try: self.connection = libvirt.openAuth(uri, auth, 0) self.last_error = None except libvirtError as e: self.last_error = 'Connection Failed: ' + str(e) self.connection = None
def _auth_callback(self, credentials, user_data): for credential in credentials: if credential[0] == libvirt.VIR_CRED_AUTHNAME: credential[4] = self.options.esx.username elif credential[0] == libvirt.VIR_CRED_NOECHOPROMPT: credential[4] = self.options.esx.password else: raise CuckooCriticalError("ESX machinery did not recieve an object to inject a username or password into") return 0
def _global_connect(self): """Set the single connection handle.""" try: self.auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_NOECHOPROMPT], self._auth_callback, None] return libvirt.openAuth(self.dsn, self.auth, 0) except libvirt.libvirtError as libvex: raise CuckooCriticalError("libvirt returned an exception on connection: %s" % libvex)
def _get_auth_conn(config): def request_cred(credentials, user_data): for credential in credentials: if credential[0] == libvirt.VIR_CRED_AUTHNAME: credential[4] = config.get("username") elif credential[0] == libvirt.VIR_CRED_PASSPHRASE: credential[4] = config.get("password") return 0 auth = [ [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE], request_cred, None ] return libvirt.openAuth(config["uri"], auth, 0)
def _global_connect(self): """ set the single connection handle """ try: self.auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_NOECHOPROMPT], self._auth_callback, None] return libvirt.openAuth(self.dsn, self.auth, 0) except libvirt.libvirtError as libvex: raise CuckooCriticalError("libvirt returned an exception on connection: %s" % libvex)
def request_cred(self, credentials, user_data): for credential in credentials: if credential[0] == libvirt.VIR_CRED_AUTHNAME: credential[4] = self.host_username elif credential[0] == libvirt.VIR_CRED_PASSPHRASE: credential[4] = self.host_password return 0
def connect(self): """ Returns None in case of failed connection. """ conn = None if self.conn_status is False: return False if self.host_protocol == 'libssh2': self.auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE], self.request_cred, None] if self.host_url == None: conn = libvirt.open("qemu:///system") else: url = "{}://{}/?sshauth=password".format(VIRT_CONN_LIBSSH2, self.host_url) conn = libvirt.openAuth(url, self.auth, 0) elif self.host_protocol == 'ssh': if self.host_url == None: conn = libvirt.open("qemu:///system") else: url = "{}://{}@{}/system?socket=/var/run/libvirt/libvirt-sock&keyfile={}".format(VIRT_CONN_SSH, self.host_username, self.host_url, self.host_key) conn = libvirt.open(url) elif self.host_protocol == 'qemu': conn = libvirt.open("qemu:///system") if conn == None: logging.error('Connection to hypervisor failed!') return False else: logging.info('Connection succesfull.') self.conn = conn SETTINGS['connection'] = self.conn
def __connect(self): """This function establishes a connection to the hypervisor.""" #create weirdo auth dict auth = [ [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE], self.retrieve_credentials, None ] #authenticate self.SESSION = libvirt.openAuth(self.URI, auth, 0) if self.SESSION == None: raise SessionException("Unable to establish connection to hypervisor!")
def test_open_auth_ignored(self): def req(credentials, user_data): for cred in credentials: if cred[0] == libvirt.VIR_CRED_AUTHNAME: cred[4] = 'convirt' elif cred[0] == libvirt.VIR_CRED_PASSPHRASE: cred[4] = 'ignored' return 0 auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE], req, None] conn = convirt.openAuth('convirt:///system', auth) self.assertTrue(conn)
def auth_callback(credentials, user_data): for credential in credentials: if credential[0] == libvirt.VIR_CRED_AUTHNAME: credential[4] = config.get('libvirt_username') elif credential[0] == libvirt.VIR_CRED_PASSPHRASE: credential[4] = config.get('libvirt_password') return 0
def get_libvirt_connection(name, libvirt_url='qemu:///system'): if name not in LIBVIRT_CONNECTIONS: auth = [ [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE], auth_callback, None ] LIBVIRT_CONNECTIONS[name] = libvirt.openAuth(libvirt_url, auth) return LIBVIRT_CONNECTIONS[name]
def __connect_tcp(self): flags = [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE] auth = [flags, self.__libvirt_auth_credentials_callback, None] uri = 'qemu+tcp://%s/system' % self.host try: self.connection = libvirt.openAuth(uri, auth, 0) self.last_error = None except libvirtError as e: self.last_error = 'Connection Failed: ' + str(e) log_error(self.last_error) self.connection = None
