Coverage for /pythoncovmergedfiles/medio/medio/src/paramiko/paramiko/auth_handler.py: 12%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright (C) 2003-2007 Robey Pointer <robeypointer@gmail.com>
2#
3# This file is part of paramiko.
4#
5# Paramiko is free software; you can redistribute it and/or modify it under the
6# terms of the GNU Lesser General Public License as published by the Free
7# Software Foundation; either version 2.1 of the License, or (at your option)
8# any later version.
9#
10# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY
11# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
12# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
13# details.
14#
15# You should have received a copy of the GNU Lesser General Public License
16# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
17# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
19"""
20`.AuthHandler`
21"""
23import threading
24import time
25import weakref
27from paramiko.common import (
28 AUTH_FAILED,
29 AUTH_PARTIALLY_SUCCESSFUL,
30 AUTH_SUCCESSFUL,
31 DEBUG,
32 DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE,
33 DISCONNECT_SERVICE_NOT_AVAILABLE,
34 INFO,
35 MSG_SERVICE_ACCEPT,
36 MSG_SERVICE_REQUEST,
37 MSG_USERAUTH_BANNER,
38 MSG_USERAUTH_FAILURE,
39 MSG_USERAUTH_INFO_REQUEST,
40 MSG_USERAUTH_INFO_RESPONSE,
41 MSG_USERAUTH_REQUEST,
42 MSG_USERAUTH_SUCCESS,
43 WARNING,
44 cMSG_DISCONNECT,
45 cMSG_SERVICE_ACCEPT,
46 cMSG_SERVICE_REQUEST,
47 cMSG_USERAUTH_BANNER,
48 cMSG_USERAUTH_FAILURE,
49 cMSG_USERAUTH_INFO_REQUEST,
50 cMSG_USERAUTH_INFO_RESPONSE,
51 cMSG_USERAUTH_PK_OK,
52 cMSG_USERAUTH_REQUEST,
53 cMSG_USERAUTH_SUCCESS,
54)
55from paramiko.message import Message
56from paramiko.server import InteractiveQuery
57from paramiko.ssh_exception import (
58 AuthenticationException,
59 BadAuthenticationType,
60 PartialAuthentication,
61 SSHException,
62)
63from paramiko.util import b, u
66class AuthHandler:
67 """
68 Internal class to handle the mechanics of authentication.
69 """
71 def __init__(self, transport):
72 self.transport = weakref.proxy(transport)
73 self.username = None
74 self.authenticated = False
75 self.auth_event = None
76 self.auth_method = ""
77 self.banner = None
78 self.password = None
79 self.private_key = None
80 self.interactive_handler = None
81 self.submethods = None
82 # for server mode:
83 self.auth_username = None
84 self.auth_fail_count = 0
86 def _log(self, *args):
87 return self.transport._log(*args)
89 def is_authenticated(self):
90 return self.authenticated
92 def get_username(self):
93 if self.transport.server_mode:
94 return self.auth_username
95 else:
96 return self.username
98 def auth_none(self, username, event):
99 self.transport.lock.acquire()
100 try:
101 self.auth_event = event
102 self.auth_method = "none"
103 self.username = username
104 self._request_auth()
105 finally:
106 self.transport.lock.release()
108 def auth_publickey(self, username, key, event):
109 self.transport.lock.acquire()
110 try:
111 self.auth_event = event
112 self.auth_method = "publickey"
113 self.username = username
114 self.private_key = key
115 self._request_auth()
116 finally:
117 self.transport.lock.release()
119 def auth_password(self, username, password, event):
120 self.transport.lock.acquire()
121 try:
122 self.auth_event = event
123 self.auth_method = "password"
124 self.username = username
125 self.password = password
126 self._request_auth()
127 finally:
128 self.transport.lock.release()
130 def auth_interactive(self, username, handler, event, submethods=""):
131 """
132 response_list = handler(title, instructions, prompt_list)
133 """
134 self.transport.lock.acquire()
135 try:
136 self.auth_event = event
137 self.auth_method = "keyboard-interactive"
138 self.username = username
139 self.interactive_handler = handler
140 self.submethods = submethods
141 self._request_auth()
142 finally:
143 self.transport.lock.release()
145 def abort(self):
146 if self.auth_event is not None:
147 self.auth_event.set()
149 # ...internals...
151 def _request_auth(self):
152 m = Message()
153 m.add_byte(cMSG_SERVICE_REQUEST)
154 m.add_string("ssh-userauth")
155 self.transport._send_message(m)
157 def _disconnect_service_not_available(self):
158 m = Message()
159 m.add_byte(cMSG_DISCONNECT)
160 m.add_int(DISCONNECT_SERVICE_NOT_AVAILABLE)
161 m.add_string("Service not available")
162 m.add_string("en")
163 self.transport._send_message(m)
164 self.transport.close()
166 def _disconnect_no_more_auth(self):
167 m = Message()
168 m.add_byte(cMSG_DISCONNECT)
169 m.add_int(DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE)
170 m.add_string("No more auth methods available")
171 m.add_string("en")
172 self.transport._send_message(m)
173 self.transport.close()
175 def _get_key_type_and_bits(self, key):
176 """
177 Given any key, return its type/algorithm & bits-to-sign.
179 Intended for input to or verification of, key signatures.
180 """
181 # Use certificate contents, if available, plain pubkey otherwise
182 if key.public_blob:
183 return key.public_blob.key_type, key.public_blob.key_blob
184 else:
185 return key.get_name(), key
187 def _get_session_blob(self, key, service, username, algorithm):
188 m = Message()
189 m.add_string(self.transport.session_id)
190 m.add_byte(cMSG_USERAUTH_REQUEST)
191 m.add_string(username)
192 m.add_string(service)
193 m.add_string("publickey")
194 m.add_boolean(True)
195 _, bits = self._get_key_type_and_bits(key)
196 m.add_string(algorithm)
197 m.add_string(bits)
198 return m.asbytes()
200 def wait_for_response(self, event):
201 max_ts = None
202 if self.transport.auth_timeout is not None:
203 max_ts = time.time() + self.transport.auth_timeout
204 while True:
205 event.wait(0.1)
206 if not self.transport.is_active():
207 e = self.transport.get_exception()
208 if (e is None) or issubclass(e.__class__, EOFError):
209 e = AuthenticationException(
210 "Authentication failed: transport shut down or saw EOF"
211 )
212 raise e
213 if event.is_set():
214 break
215 if max_ts is not None and max_ts <= time.time():
216 raise AuthenticationException("Authentication timeout.")
218 if not self.is_authenticated():
219 e = self.transport.get_exception()
220 if e is None:
221 e = AuthenticationException("Authentication failed.")
222 # this is horrible. Python Exception isn't yet descended from
223 # object, so type(e) won't work. :(
224 # TODO (backwards incompat): lol. just lmao.
225 if issubclass(e.__class__, PartialAuthentication):
226 return e.allowed_types
227 raise e
228 return []
230 def _parse_service_request(self, m):
231 service = m.get_text()
232 if self.transport.server_mode and (service == "ssh-userauth"):
233 # accepted
234 m = Message()
235 m.add_byte(cMSG_SERVICE_ACCEPT)
236 m.add_string(service)
237 self.transport._send_message(m)
238 banner, language = self.transport.server_object.get_banner()
239 if banner:
240 m = Message()
241 m.add_byte(cMSG_USERAUTH_BANNER)
242 m.add_string(banner)
243 m.add_string(language)
244 self.transport._send_message(m)
245 return
246 # dunno this one
247 self._disconnect_service_not_available()
249 def _generate_key_from_request(self, algorithm, keyblob):
250 # For use in server mode.
251 options = self.transport.preferred_pubkeys
252 if algorithm.replace("-cert-v01@openssh.com", "") not in options:
253 err = (
254 "Auth rejected: pubkey algorithm '{}' unsupported or disabled"
255 )
256 self._log(INFO, err.format(algorithm))
257 return None
258 return self.transport._key_info[algorithm](Message(keyblob))
260 def _finalize_pubkey_algorithm(self, key_type):
261 """
262 Given a key type, decide which pubkey algorithm to use with it.
264 In most cases this is simply "that key type, again".
266 For RSA, this will be one of the SHA2 algorithms, depending on our
267 (transport's) configured pubkey algorithms list.
268 """
269 # Short-circuit for non-RSA keys
270 if "rsa" not in key_type:
271 return key_type
272 self._log(
273 DEBUG,
274 "Finalizing pubkey algorithm for key of type {!r}".format(
275 key_type
276 ),
277 )
278 # Normal attempts to handshake follow from here.
279 # Only consider RSA algos from our list, lest we agree on another!
280 my_algos = [x for x in self.transport.preferred_pubkeys if "rsa" in x]
281 self._log(DEBUG, "Our pubkey algorithm list: {}".format(my_algos))
282 # Short-circuit negatively if user disabled all RSA algos (heh)
283 if not my_algos:
284 raise SSHException(
285 "An RSA key was specified, but no RSA pubkey algorithms are configured!" # noqa
286 )
287 # Check for server-sig-algs if supported & sent
288 server_algo_str = u(
289 self.transport.server_extensions.get("server-sig-algs", b(""))
290 )
291 pubkey_algo = None
292 # Prefer to match against server-sig-algs
293 if server_algo_str:
294 server_algos = server_algo_str.split(",")
295 self._log(
296 DEBUG, "Server-side algorithm list: {}".format(server_algos)
297 )
298 # Only use algos from our list that the server likes, in our own
299 # preference order. (NOTE: purposefully using same style as in
300 # Transport...expect to refactor later)
301 agreement = list(filter(server_algos.__contains__, my_algos))
302 if agreement:
303 pubkey_algo = agreement[0]
304 self._log(
305 DEBUG,
306 "Agreed upon {!r} pubkey algorithm".format(pubkey_algo),
307 )
308 else:
309 self._log(DEBUG, "No common pubkey algorithms exist! Dying.")
310 # TODO: MAY want to use IncompatiblePeer again here but that's
311 # technically for initial key exchange, not pubkey auth.
312 err = "Unable to agree on a pubkey algorithm for signing a {!r} key!" # noqa
313 raise AuthenticationException(err.format(key_type))
314 # Fallback to first item in our preferred algorithm list (which won't
315 # be empty due to guardrail above)
316 else:
317 pubkey_algo = my_algos[0]
318 msg = f"Server did not send a server-sig-algs list; defaulting to first RSA algorithm in our list: {pubkey_algo}" # noqa
319 self._log(DEBUG, msg)
320 # If we had loaded a cert-type key, tack that on to the algorithm name
321 # to get the final correct result.
322 if key_type.endswith("-cert-v01@openssh.com"):
323 pubkey_algo += "-cert-v01@openssh.com"
324 self.transport._agreed_pubkey_algorithm = pubkey_algo
325 return pubkey_algo
327 def _parse_service_accept(self, m):
328 service = m.get_text()
329 if service == "ssh-userauth":
330 self._log(DEBUG, "userauth is OK")
331 m = Message()
332 m.add_byte(cMSG_USERAUTH_REQUEST)
333 m.add_string(self.username)
334 m.add_string("ssh-connection")
335 m.add_string(self.auth_method)
336 if self.auth_method == "password":
337 m.add_boolean(False)
338 password = b(self.password)
339 m.add_string(password)
340 elif self.auth_method == "publickey":
341 m.add_boolean(True)
342 key_type, bits = self._get_key_type_and_bits(self.private_key)
343 algorithm = self._finalize_pubkey_algorithm(key_type)
344 m.add_string(algorithm)
345 m.add_string(bits)
346 blob = self._get_session_blob(
347 self.private_key,
348 "ssh-connection",
349 self.username,
350 algorithm,
351 )
352 sig = self.private_key.sign_ssh_data(blob, algorithm)
353 m.add_string(sig)
354 elif self.auth_method == "keyboard-interactive":
355 m.add_string("")
356 m.add_string(self.submethods)
357 elif self.auth_method == "none":
358 pass
359 else:
360 raise SSHException(
361 'Unknown auth method "{}"'.format(self.auth_method)
362 )
363 self.transport._send_message(m)
364 else:
365 self._log(
366 DEBUG, 'Service request "{}" accepted (?)'.format(service)
367 )
369 def _send_auth_result(self, username, method, result):
370 # okay, send result
371 m = Message()
372 if result == AUTH_SUCCESSFUL:
373 self._log(INFO, "Auth granted ({}).".format(method))
374 m.add_byte(cMSG_USERAUTH_SUCCESS)
375 self.authenticated = True
376 else:
377 self._log(INFO, "Auth rejected ({}).".format(method))
378 m.add_byte(cMSG_USERAUTH_FAILURE)
379 m.add_string(
380 self.transport.server_object.get_allowed_auths(username)
381 )
382 if result == AUTH_PARTIALLY_SUCCESSFUL:
383 m.add_boolean(True)
384 else:
385 m.add_boolean(False)
386 self.auth_fail_count += 1
387 self.transport._send_message(m)
388 if self.auth_fail_count >= 10:
389 self._disconnect_no_more_auth()
390 if result == AUTH_SUCCESSFUL:
391 self.transport._auth_trigger()
393 def _interactive_query(self, q):
394 # make interactive query instead of response
395 m = Message()
396 m.add_byte(cMSG_USERAUTH_INFO_REQUEST)
397 m.add_string(q.name)
398 m.add_string(q.instructions)
399 m.add_string(bytes())
400 m.add_int(len(q.prompts))
401 for p in q.prompts:
402 m.add_string(p[0])
403 m.add_boolean(p[1])
404 self.transport._send_message(m)
406 def _parse_userauth_request(self, m):
407 if not self.transport.server_mode:
408 # er, uh... what?
409 m = Message()
410 m.add_byte(cMSG_USERAUTH_FAILURE)
411 m.add_string("none")
412 m.add_boolean(False)
413 self.transport._send_message(m)
414 return
415 if self.authenticated:
416 # ignore
417 return
418 username = m.get_text()
419 service = m.get_text()
420 method = m.get_text()
421 self._log(
422 DEBUG,
423 "Auth request (type={}) service={}, username={}".format(
424 method, service, username
425 ),
426 )
427 if service != "ssh-connection":
428 self._disconnect_service_not_available()
429 return
430 if (self.auth_username is not None) and (
431 self.auth_username != username
432 ):
433 self._log(
434 WARNING,
435 "Auth rejected because the client attempted to change username in mid-flight", # noqa
436 )
437 self._disconnect_no_more_auth()
438 return
439 self.auth_username = username
441 if method == "none":
442 result = self.transport.server_object.check_auth_none(username)
443 elif method == "password":
444 changereq = m.get_boolean()
445 password = m.get_binary()
446 try:
447 password = password.decode("UTF-8")
448 except UnicodeError:
449 # some clients/servers expect non-utf-8 passwords!
450 # in this case, just return the raw byte string.
451 pass
452 if changereq:
453 # always treated as failure, since we don't support changing
454 # passwords, but collect the list of valid auth types from
455 # the callback anyway
456 self._log(DEBUG, "Auth request to change passwords (rejected)")
457 newpassword = m.get_binary()
458 try:
459 newpassword = newpassword.decode("UTF-8", "replace")
460 except UnicodeError:
461 pass
462 result = AUTH_FAILED
463 else:
464 result = self.transport.server_object.check_auth_password(
465 username, password
466 )
467 elif method == "publickey":
468 sig_attached = m.get_boolean()
469 # NOTE: server never wants to guess a client's algo, they're
470 # telling us directly. No need for _finalize_pubkey_algorithm
471 # anywhere in this flow.
472 # TODO: ok is this a spot where it can say a SHA2 dealie in some
473 # fields but still ssh-rsa within the pubkey blob part?
474 # TODO: ok so this would be rsa-sha2-256 or w/e, if this field says
475 # ssh-rsa the request can get stuffed.
476 algorithm = m.get_text()
477 # TODO: This part would, if deconstructed, still be allowed to have
478 # "ssh-rsa" in its first field.
479 keyblob = m.get_binary()
480 try:
481 key = self._generate_key_from_request(algorithm, keyblob)
482 except SSHException as e:
483 self._log(INFO, "Auth rejected: public key: {}".format(str(e)))
484 key = None
485 except Exception as e:
486 msg = "Auth rejected: unsupported or mangled public key ({}: {})" # noqa
487 self._log(INFO, msg.format(e.__class__.__name__, e))
488 key = None
489 if key is None:
490 self._disconnect_no_more_auth()
491 return
492 # first check if this key is okay... if not, we can skip the verify
493 result = self.transport.server_object.check_auth_publickey(
494 username, key
495 )
496 if result != AUTH_FAILED:
497 # key is okay, verify it
498 if not sig_attached:
499 # client wants to know if this key is acceptable, before it
500 # signs anything... send special "ok" message
501 m = Message()
502 m.add_byte(cMSG_USERAUTH_PK_OK)
503 m.add_string(algorithm)
504 m.add_string(keyblob)
505 self.transport._send_message(m)
506 return
507 sig = Message(m.get_binary())
508 blob = self._get_session_blob(
509 key, service, username, algorithm
510 )
511 if not key.verify_ssh_sig(blob, sig):
512 self._log(INFO, "Auth rejected: invalid signature")
513 result = AUTH_FAILED
514 elif method == "keyboard-interactive":
515 submethods = m.get_string()
516 result = self.transport.server_object.check_auth_interactive(
517 username, submethods
518 )
519 if isinstance(result, InteractiveQuery):
520 # make interactive query instead of response
521 self._interactive_query(result)
522 return
523 else:
524 result = self.transport.server_object.check_auth_none(username)
525 # okay, send result
526 self._send_auth_result(username, method, result)
528 def _parse_userauth_success(self, m):
529 self._log(
530 INFO, "Authentication ({}) successful!".format(self.auth_method)
531 )
532 self.authenticated = True
533 self.transport._auth_trigger()
534 if self.auth_event is not None:
535 self.auth_event.set()
537 def _parse_userauth_failure(self, m):
538 authlist = m.get_list()
539 # TODO (backwards incompat): we aren't giving callers access to
540 # authlist _unless_ it's partial authentication, so eg authtype=none
541 # can't work unless we tweak this.
542 partial = m.get_boolean()
543 if partial:
544 self._log(INFO, "Authentication continues...")
545 self._log(DEBUG, "Methods: " + str(authlist))
546 self.transport.saved_exception = PartialAuthentication(authlist)
547 elif self.auth_method not in authlist:
548 for msg in (
549 "Authentication type ({}) not permitted.".format(
550 self.auth_method
551 ),
552 "Allowed methods: {}".format(authlist),
553 ):
554 self._log(DEBUG, msg)
555 self.transport.saved_exception = BadAuthenticationType(
556 "Bad authentication type", authlist
557 )
558 else:
559 self._log(
560 INFO, "Authentication ({}) failed.".format(self.auth_method)
561 )
562 self.authenticated = False
563 self.username = None
564 if self.auth_event is not None:
565 self.auth_event.set()
567 def _parse_userauth_banner(self, m):
568 banner = m.get_string()
569 self.banner = banner
570 self._log(INFO, "Auth banner: {}".format(banner))
571 # who cares.
573 def _parse_userauth_info_request(self, m):
574 if self.auth_method != "keyboard-interactive":
575 raise SSHException("Illegal info request from server")
576 title = m.get_text()
577 instructions = m.get_text()
578 m.get_binary() # lang
579 prompts = m.get_int()
580 prompt_list = []
581 for i in range(prompts):
582 prompt_list.append((m.get_text(), m.get_boolean()))
583 response_list = self.interactive_handler(
584 title, instructions, prompt_list
585 )
587 m = Message()
588 m.add_byte(cMSG_USERAUTH_INFO_RESPONSE)
589 m.add_int(len(response_list))
590 for r in response_list:
591 m.add_string(r)
592 self.transport._send_message(m)
594 def _parse_userauth_info_response(self, m):
595 if not self.transport.server_mode:
596 raise SSHException("Illegal info response from server")
597 n = m.get_int()
598 responses = []
599 for i in range(n):
600 responses.append(m.get_text())
601 result = self.transport.server_object.check_auth_interactive_response(
602 responses
603 )
604 if isinstance(result, InteractiveQuery):
605 # make interactive query instead of response
606 self._interactive_query(result)
607 return
608 self._send_auth_result(
609 self.auth_username, "keyboard-interactive", result
610 )
612 # TODO (backwards incompat): MAY make sense to make these tables into
613 # actual classes/instances that can be fed a mode bool or whatever. Or,
614 # alternately (both?) make the message types small classes or enums that
615 # embed this info within themselves (which could also then tidy up the
616 # current 'integer -> human readable short string' stuff in common.py).
617 # TODO: if we do that, also expose 'em publicly.
619 # Messages which should be handled _by_ servers (sent by clients)
620 @property
621 def _server_handler_table(self):
622 return {
623 # TODO (backwards incompat): MSG_SERVICE_REQUEST ought to
624 # eventually move into Transport's server mode like the client side
625 # did, just for consistency.
626 MSG_SERVICE_REQUEST: self._parse_service_request,
627 MSG_USERAUTH_REQUEST: self._parse_userauth_request,
628 MSG_USERAUTH_INFO_RESPONSE: self._parse_userauth_info_response,
629 }
631 # Messages which should be handled _by_ clients (sent by servers)
632 @property
633 def _client_handler_table(self):
634 return {
635 MSG_SERVICE_ACCEPT: self._parse_service_accept,
636 MSG_USERAUTH_SUCCESS: self._parse_userauth_success,
637 MSG_USERAUTH_FAILURE: self._parse_userauth_failure,
638 MSG_USERAUTH_BANNER: self._parse_userauth_banner,
639 MSG_USERAUTH_INFO_REQUEST: self._parse_userauth_info_request,
640 }
642 # NOTE: prior to the fix for #1283, this was a static dict instead of a
643 # property. Should be backwards compatible in most/all cases.
644 @property
645 def _handler_table(self):
646 if self.transport.server_mode:
647 return self._server_handler_table
648 else:
649 return self._client_handler_table
652class AuthOnlyHandler(AuthHandler):
653 """
654 AuthHandler, and just auth, no service requests!
656 .. versionadded:: 3.2
657 """
659 # NOTE: this purposefully duplicates some of the parent class in order to
660 # modernize, refactor, etc. The intent is that eventually we will collapse
661 # this one onto the parent in a backwards incompatible release.
663 @property
664 def _client_handler_table(self):
665 my_table = super()._client_handler_table.copy()
666 del my_table[MSG_SERVICE_ACCEPT]
667 return my_table
669 def send_auth_request(self, username, method, finish_message=None):
670 """
671 Submit a userauth request message & wait for response.
673 Performs the transport message send call, sets self.auth_event, and
674 will lock-n-block as necessary to both send, and wait for response to,
675 the USERAUTH_REQUEST.
677 Most callers will want to supply a callback to ``finish_message``,
678 which accepts a Message ``m`` and may call mutator methods on it to add
679 more fields.
680 """
681 # Store a few things for reference in handlers, including auth failure
682 # handler (which needs to know if we were using a bad method, etc)
683 self.auth_method = method
684 self.username = username
685 # Generic userauth request fields
686 m = Message()
687 m.add_byte(cMSG_USERAUTH_REQUEST)
688 m.add_string(username)
689 m.add_string("ssh-connection")
690 m.add_string(method)
691 # Caller usually has more to say, such as injecting password, key etc
692 finish_message(m)
693 # TODO (backwards incompat): seems odd to have the client handle the
694 # lock and not Transport; that _may_ have been an artifact of allowing
695 # user threading event injection? Regardless, we don't want to move
696 # _this_ locking into Transport._send_message now, because lots of
697 # other untouched code also uses that method and we might end up
698 # double-locking (?) but 4.0 would be a good time to revisit.
699 with self.transport.lock:
700 self.transport._send_message(m)
701 # We have cut out the higher level event args, but self.auth_event is
702 # still required for self.wait_for_response to function correctly (it's
703 # the mechanism used by the auth success/failure handlers, the abort
704 # handler, and a few other spots.
705 # TODO: interestingly, wait_for_response itself doesn't actually
706 # enforce that its event argument and self.auth_event are the same...
707 self.auth_event = threading.Event()
708 return self.wait_for_response(self.auth_event)
710 def auth_none(self, username):
711 return self.send_auth_request(username, "none")
713 def auth_publickey(self, username, key):
714 # NOTE: key_type here may be just key type ("rsa-sha2-256") or may be
715 # the cert form if cert was loaded ("ssh-rsa-cert-v01@openssh.com")
716 key_type, bits = self._get_key_type_and_bits(key)
717 algorithm = self._finalize_pubkey_algorithm(key_type)
718 blob = self._get_session_blob(
719 key,
720 "ssh-connection",
721 username,
722 algorithm,
723 )
725 def finish(m):
726 # This field doesn't appear to be named, but is False when querying
727 # for permission (ie knowing whether to even prompt a user for
728 # passphrase, etc) or True when just going for it. Paramiko has
729 # never bothered with the former type of message, apparently.
730 m.add_boolean(True)
731 m.add_string(algorithm)
732 m.add_string(bits)
733 m.add_string(key.sign_ssh_data(blob, algorithm))
735 return self.send_auth_request(username, "publickey", finish)
737 def auth_password(self, username, password):
738 def finish(m):
739 # Unnamed field that equates to "I am changing my password", which
740 # Paramiko clientside never supported and serverside only sort of
741 # supported.
742 m.add_boolean(False)
743 m.add_string(b(password))
745 return self.send_auth_request(username, "password", finish)
747 def auth_interactive(self, username, handler, submethods=""):
748 """
749 response_list = handler(title, instructions, prompt_list)
750 """
751 # Unlike most siblings, this auth method _does_ require other
752 # superclass handlers (eg userauth info request) to understand
753 # what's going on, so we still set some self attributes.
754 self.auth_method = "keyboard_interactive"
755 self.interactive_handler = handler
757 def finish(m):
758 # Empty string for deprecated language tag field, per RFC 4256:
759 # https://www.rfc-editor.org/rfc/rfc4256#section-3.1
760 m.add_string("")
761 m.add_string(submethods)
763 return self.send_auth_request(username, "keyboard-interactive", finish)