mirror of
https://git.tartarus.org/simon/putty.git
synced 2025-01-09 09:27:59 +00:00
329a4cdd79
Python 3's stderr was fully-buffered when non-interactive, unlike Python 2 and more or less everything else, until 3.9 in 2020(!): https://bugs.python.org/issue13601 (It would be less faff to sys.stderr.reconfigure(line_buffering=True) at the start, but that was only added in 3.7, whereas the 'flush' argument to print() dates back to 3.3, so I chose that to minimise the risk of version dependencies getting in the way of using this as a working example.)
291 lines
10 KiB
Python
Executable File
291 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
# This is a demonstration example of how to write a
|
|
# keyboard-interactive authentication helper plugin using PuTTY's
|
|
# protocol for involving it in SSH connection setup.
|
|
# The protocol, and the purpose of an authentication plugin, is
|
|
# fully documented in an appendix to the PuTTY manual.
|
|
|
|
import io
|
|
import os
|
|
import struct
|
|
import sys
|
|
|
|
# Exception class we'll use to get a clean exit on EOF.
|
|
class PluginEOF(Exception): pass
|
|
|
|
# ----------------------------------------------------------------------
|
|
#
|
|
# Marshalling and unmarshalling routines to write and read the
|
|
# necessary SSH data types to/from a binary file handle (which can
|
|
# include an io.BytesIO if you need to encode/decode in-process).
|
|
#
|
|
# Error handling is a totally ad-hoc mixture of 'assert' and just
|
|
# assuming things will have the right type, or be the right length of
|
|
# tuple, or be valid UTF-8. So it should be _robust_, in the sense
|
|
# that you'll get a Python exception if anything fails. But no
|
|
# sensible error reporting or recovery is implemented.
|
|
#
|
|
# That should be good enough, because PuTTY will log the plugin's
|
|
# standard error in its Event Log, so if the plugin crashes, you'll be
|
|
# able to retrieve the traceback.
|
|
|
|
def wr_byte(fh, b):
|
|
assert 0 <= b < 0x100
|
|
fh.write(bytes([b]))
|
|
|
|
def wr_boolean(fh, b):
|
|
wr_byte(fh, 1 if b else 0)
|
|
|
|
def wr_uint32(fh, u):
|
|
assert 0 <= u < 0x100000000
|
|
fh.write(struct.pack(">I", u))
|
|
|
|
def wr_string(fh, s):
|
|
wr_uint32(fh, len(s))
|
|
fh.write(s)
|
|
|
|
def wr_string_utf8(fh, s):
|
|
wr_string(fh, s.encode("UTF-8"))
|
|
|
|
def rd_n(fh, n):
|
|
data = fh.read(n)
|
|
if len(data) < n:
|
|
raise PluginEOF()
|
|
return data
|
|
|
|
def rd_byte(fh):
|
|
return rd_n(fh, 1)[0]
|
|
|
|
def rd_boolean(fh):
|
|
return rd_byte(fh) != 0
|
|
|
|
def rd_uint32(fh):
|
|
return struct.unpack(">I", rd_n(fh, 4))[0]
|
|
|
|
def rd_string(fh):
|
|
length = rd_uint32(fh)
|
|
return rd_n(fh, length)
|
|
|
|
def rd_string_utf8(fh):
|
|
return rd_string(fh).decode("UTF-8")
|
|
|
|
# ----------------------------------------------------------------------
|
|
#
|
|
# Protocol definitions.
|
|
|
|
our_max_version = 2
|
|
|
|
PLUGIN_INIT = 1
|
|
PLUGIN_INIT_RESPONSE = 2
|
|
PLUGIN_PROTOCOL = 3
|
|
PLUGIN_PROTOCOL_ACCEPT = 4
|
|
PLUGIN_PROTOCOL_REJECT = 5
|
|
PLUGIN_AUTH_SUCCESS = 6
|
|
PLUGIN_AUTH_FAILURE = 7
|
|
PLUGIN_INIT_FAILURE = 8
|
|
PLUGIN_KI_SERVER_REQUEST = 20
|
|
PLUGIN_KI_SERVER_RESPONSE = 21
|
|
PLUGIN_KI_USER_REQUEST = 22
|
|
PLUGIN_KI_USER_RESPONSE = 23
|
|
|
|
# ----------------------------------------------------------------------
|
|
#
|
|
# Classes to make it easy to construct and receive messages.
|
|
#
|
|
# OutMessage is constructed with the message type; then you use the
|
|
# wr_foo() routines to add fields to it, and finally call its send()
|
|
# method.
|
|
#
|
|
# InMessage is constructed via the expect() class method, to which you
|
|
# give a list of message types you expect to see one of at this stage.
|
|
# Once you've got one, you can rd_foo() fields from it.
|
|
|
|
class OutMessage:
|
|
def __init__(self, msgtype):
|
|
self.buf = io.BytesIO()
|
|
wr_byte(self.buf, msgtype)
|
|
self.write = self.buf.write
|
|
|
|
def send(self, fh=sys.stdout.buffer):
|
|
wr_string(fh, self.buf.getvalue())
|
|
fh.flush()
|
|
|
|
class InMessage:
|
|
@classmethod
|
|
def expect(cls, expected_types, fh=sys.stdin.buffer):
|
|
self = cls()
|
|
self.buf = io.BytesIO(rd_string(fh))
|
|
self.msgtype = rd_byte(self.buf)
|
|
self.read = self.buf.read
|
|
|
|
if self.msgtype not in expected_types:
|
|
raise ValueError("received packet type {:d}, expected {}".format(
|
|
self.msgtype, ",".join(map("{:d}".format,
|
|
sorted(expected_types)))))
|
|
return self
|
|
|
|
# ----------------------------------------------------------------------
|
|
#
|
|
# The main implementation of the protocol.
|
|
|
|
def protocol():
|
|
# Start by expecting PLUGIN_INIT.
|
|
msg = InMessage.expect({PLUGIN_INIT})
|
|
their_version = rd_uint32(msg)
|
|
hostname = rd_string_utf8(msg)
|
|
port = rd_uint32(msg)
|
|
username = rd_string_utf8(msg)
|
|
print(f"Got hostname {hostname!r}, port {port!r}", file=sys.stderr,
|
|
flush=True)
|
|
|
|
# Decide which protocol version we're speaking.
|
|
version = min(their_version, our_max_version)
|
|
assert version != 0, "Protocol version 0 does not exist"
|
|
|
|
if "TESTPLUGIN_INIT_FAIL" in os.environ:
|
|
# Test the plugin failing at startup time.
|
|
msg = OutMessage(PLUGIN_INIT_FAILURE)
|
|
wr_string_utf8(msg, os.environ["TESTPLUGIN_INIT_FAIL"])
|
|
msg.send()
|
|
return
|
|
|
|
# Send INIT_RESPONSE, with our protocol version and an overridden
|
|
# username.
|
|
#
|
|
# By default this test plugin doesn't override the username, but
|
|
# you can make it do so by setting TESTPLUGIN_USERNAME in the
|
|
# environment.
|
|
msg = OutMessage(PLUGIN_INIT_RESPONSE)
|
|
wr_uint32(msg, version)
|
|
wr_string_utf8(msg, os.environ.get("TESTPLUGIN_USERNAME", ""))
|
|
msg.send()
|
|
|
|
# Outer loop run once per authentication protocol.
|
|
while True:
|
|
# Expect a message telling us what the protocol is.
|
|
msg = InMessage.expect({PLUGIN_PROTOCOL})
|
|
method = rd_string(msg)
|
|
|
|
if "TESTPLUGIN_PROTO_REJECT" in os.environ:
|
|
# Test the plugin failing at PLUGIN_PROTOCOL time.
|
|
msg = OutMessage(PLUGIN_PROTOCOL_REJECT)
|
|
wr_string_utf8(msg, os.environ["TESTPLUGIN_PROTO_REJECT"])
|
|
msg.send()
|
|
continue
|
|
|
|
# We only support keyboard-interactive. If we supported other
|
|
# auth methods, this would be the place to add further clauses
|
|
# to this if statement for them.
|
|
if method == b"keyboard-interactive":
|
|
msg = OutMessage(PLUGIN_PROTOCOL_ACCEPT)
|
|
msg.send()
|
|
|
|
# Inner loop run once per keyboard-interactive exchange
|
|
# with the SSH server.
|
|
while True:
|
|
# Expect a set of prompts from the server, or
|
|
# terminate the loop on SUCCESS or FAILURE.
|
|
#
|
|
# (We could also respond to SUCCESS or FAILURE by
|
|
# updating caches of our own, if we had any that were
|
|
# useful.)
|
|
msg = InMessage.expect({PLUGIN_KI_SERVER_REQUEST,
|
|
PLUGIN_AUTH_SUCCESS,
|
|
PLUGIN_AUTH_FAILURE})
|
|
if (msg.msgtype == PLUGIN_AUTH_SUCCESS or
|
|
msg.msgtype == PLUGIN_AUTH_FAILURE):
|
|
break
|
|
|
|
# If we didn't just break, we're sitting on a
|
|
# PLUGIN_KI_SERVER_REQUEST message. Get all its bits
|
|
# and pieces out.
|
|
name = rd_string_utf8(msg)
|
|
instructions = rd_string_utf8(msg)
|
|
language = rd_string(msg)
|
|
nprompts = rd_uint32(msg)
|
|
prompts = []
|
|
for i in range(nprompts):
|
|
prompt = rd_string_utf8(msg)
|
|
echo = rd_boolean(msg)
|
|
prompts.append((prompt, echo))
|
|
|
|
# Actually make up some answers for the prompts. This
|
|
# is the part that a non-example implementation would
|
|
# do very differently, of course!
|
|
#
|
|
# Here, we answer "foo" to every prompt, except that
|
|
# if there are exactly two prompts in the packet then
|
|
# we answer "stoat" to the first and "weasel" to the
|
|
# second.
|
|
#
|
|
# (These answers are consistent with the ones required
|
|
# by PuTTY's test SSH server Uppity in its own
|
|
# keyboard-interactive test implementation: that
|
|
# presents a two-prompt packet and expects
|
|
# "stoat","weasel" as the answers, and then presents a
|
|
# zero-prompt packet. So this test plugin will get you
|
|
# through Uppity's k-i in a one-touch manner. The
|
|
# "foo" in this code isn't used by Uppity at all; I
|
|
# just include it because I had to have _some_
|
|
# handling for the else clause.)
|
|
#
|
|
# If TESTPLUGIN_PROMPTS is set in the environment, we
|
|
# ask the user questions of our own by sending them
|
|
# back to PuTTY as USER_REQUEST messages.
|
|
if nprompts == 2:
|
|
if "TESTPLUGIN_PROMPTS" in os.environ:
|
|
for i in range(2):
|
|
# Make up some questions to ask.
|
|
msg = OutMessage(PLUGIN_KI_USER_REQUEST)
|
|
wr_string_utf8(
|
|
msg, "Plugin request #{:d} (name)".format(i))
|
|
wr_string_utf8(
|
|
msg, "Plugin request #{:d} (instructions)"
|
|
.format(i))
|
|
wr_string(msg, b"")
|
|
wr_uint32(msg, 2)
|
|
wr_string_utf8(msg, "Prompt 1 of 2 (echo): ")
|
|
wr_boolean(msg, True)
|
|
wr_string_utf8(msg, "Prompt 2 of 2 (no echo): ")
|
|
wr_boolean(msg, False)
|
|
msg.send()
|
|
|
|
# Expect the answers.
|
|
msg = InMessage.expect({PLUGIN_KI_USER_RESPONSE})
|
|
user_nprompts = rd_uint32(msg)
|
|
assert user_nprompts == 2, (
|
|
"Should match what we just sent")
|
|
for i in range(nprompts):
|
|
user_response = rd_string_utf8(msg)
|
|
# We don't actually check these
|
|
# responses for anything.
|
|
|
|
answers = ["stoat", "weasel"]
|
|
|
|
else:
|
|
answers = ["foo"] * nprompts
|
|
|
|
# Send the answers to the SSH server's questions.
|
|
msg = OutMessage(PLUGIN_KI_SERVER_RESPONSE)
|
|
wr_uint32(msg, len(answers))
|
|
for answer in answers:
|
|
wr_string_utf8(msg, answer)
|
|
msg.send()
|
|
|
|
else:
|
|
# Default handler if we don't speak the offered protocol
|
|
# at all.
|
|
msg = OutMessage(PLUGIN_PROTOCOL_REJECT)
|
|
wr_string_utf8(msg, "")
|
|
msg.send()
|
|
|
|
# Demonstration write to stderr, to prove that it shows up in PuTTY's
|
|
# Event Log.
|
|
print("Hello from test plugin's stderr", file=sys.stderr, flush=True)
|
|
|
|
try:
|
|
protocol()
|
|
except PluginEOF:
|
|
pass
|