diff --git a/privacyidea_pam.py b/privacyidea_pam.py index 0725d23..af94bdf 100644 --- a/privacyidea_pam.py +++ b/privacyidea_pam.py @@ -1,5 +1,7 @@ # -*- coding: utf-8 -*- # +# 2016-03-03 Brandon Smith +# Add U2F challenge/response support # 2015-11-06 Cornelius Kölbel # Avoid SQL injections. # 2015-10-17 Cornelius Kölbel @@ -33,6 +35,7 @@ The code is tested in test_pam_module.py """ +import json import requests import syslog import sqlite3 @@ -75,6 +78,18 @@ def __init__(self, pamh, config): self.debug = config.get("debug") self.sqlfile = config.get("sqlfile", "/etc/privacyidea/pam.sqlite") + def make_request(self, data): + response = requests.post(self.URL + "/validate/check", data=data, + verify=self.sslverify) + + json_response = response.json + if callable(json_response): + syslog.syslog(syslog.LOG_DEBUG, "requests > 1.0") + json_response = json_response() + + return json_response + + def authenticate(self, password): rval = self.pamh.PAM_SYSTEM_ERR # First we try to authenticate against the sqlitedb @@ -91,13 +106,8 @@ def authenticate(self, password): "pass": password} if self.realm: data["realm"] = self.realm - response = requests.post(self.URL + "/validate/check", data=data, - verify=self.sslverify) - json_response = response.json - if callable(json_response): - syslog.syslog(syslog.LOG_DEBUG, "requests > 1.0") - json_response = json_response() + json_response = self.make_request(data) result = json_response.get("result") auth_item = json_response.get("auth_items") @@ -105,8 +115,10 @@ def authenticate(self, password): serial = detail.get("serial", "T%s" % time.time()) tokentype = detail.get("type", "unknown") if self.debug: - syslog.syslog(syslog.LOG_DEBUG, "%s: result: %s" % (__name__, - result)) + syslog.syslog(syslog.LOG_DEBUG, + "%s: result: %s" % (__name__, result)) + syslog.syslog(syslog.LOG_DEBUG, + "%s: detail: %s" % (__name__, detail)) if result.get("status"): if result.get("value"): @@ -114,7 +126,21 @@ def authenticate(self, password): save_auth_item(self.sqlfile, self.user, serial, tokentype, auth_item) else: - rval = self.pamh.PAM_AUTH_ERR + transaction_id = detail.get("transaction_id") + + if transaction_id: + attributes = detail.get("attributes", {}) + if "u2fSignRequest" in attributes: + rval = self.u2f_challenge_response( + transaction_id, detail.get("message"), + attributes) + else: + syslog.syslog(syslog.LOG_ERR, + "%s: unsupported challenge" % + __name__) + + else: + rval = self.pamh.PAM_AUTH_ERR else: syslog.syslog(syslog.LOG_ERR, "%s: %s" % (__name__, @@ -122,6 +148,72 @@ def authenticate(self, password): return rval + def u2f_challenge_response(self, transaction_id, message, attributes): + rval = self.pamh.PAM_SYSTEM_ERR + + syslog.syslog(syslog.LOG_DEBUG, "Prompting for U2F authentication") + +# In case of U2F "attributes" looks like this: +# { +# "img": "static/css/FIDO-U2F-Security-Key-444x444.png#012", +# "hideResponseInput" "1", +# "u2fSignRequest": { +# "challenge": "yji-PL1V0QELilDL3m6Lc-1yahpKZiU-z6ye5Zz2mp8", +# "version": "U2F_V2", +# "keyHandle": "fxDKTr6o8EEGWPyEyRVDvnoeA0c6v-dgvbN-6Mxc6XBmEItsw", +# "appId": "https://172.16.200.138" +# } +# } + challenge = """ +----- BEGIN U2F CHALLENGE ----- +%s +%s +%s +----- END U2F CHALLENGE -----""" % (self.URL, + json.dumps(attributes["u2fSignRequest"]), + str(message or "")) + + if bool(attributes.get("hideResponseInput", True)): + prompt_type = self.pamh.PAM_PROMPT_ECHO_OFF + else: + prompt_type = self.pamh.PAM_PROMPT_ECHO_ON + + message = self.pamh.Message(prompt_type, challenge) + response = self.pamh.conversation(message) + chal_response = json.loads(response.resp) + + data = {"user": self.user, + "transaction_id": transaction_id, + "pass": self.pamh.authtok, + "signaturedata": chal_response.get("signatureData"), + "clientdata": chal_response.get("clientData")} + if self.realm: + data["realm"] = self.realm + + json_response = self.make_request(data) + + result = json_response.get("result") + detail = json_response.get("detail") + + if self.debug: + syslog.syslog(syslog.LOG_DEBUG, + "%s: result: %s" % (__name__, result)) + syslog.syslog(syslog.LOG_DEBUG, + "%s: detail: %s" % (__name__, detail)) + + if result.get("status"): + if result.get("value"): + rval = self.pamh.PAM_SUCCESS + else: + rval = self.pamh.PAM_AUTH_ERR + else: + syslog.syslog(syslog.LOG_ERR, + "%s: %s" % (__name__, + result.get("error").get("message"))) + + return rval + + def pam_sm_authenticate(pamh, flags, argv): config = _get_config(argv) diff --git a/ssh-u2f.py b/ssh-u2f.py new file mode 100755 index 0000000..0d9215e --- /dev/null +++ b/ssh-u2f.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python +# +# -*- coding: utf-8 -*- +# +# 2016-03-03 Brandon Smith +# Initial Creation +# +# (c) Brandon Smith +# Info: http://www.privacyidea.org +# +# This code is free software; you can redistribute it and/or +# modify it under the terms of the GNU AFFERO GENERAL PUBLIC LICENSE +# License as published by the Free Software Foundation; either +# version 3 of the License, or any later version. +# +# This code is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU AFFERO GENERAL PUBLIC LICENSE for more details. +# +# You should have received a copy of the GNU Affero General Public +# License along with this program. If not, see . +# +from __future__ import print_function + +import getpass,os,re,signal,subprocess,sys +import pexpect + +__doc__ = """This is an ssh (and ssh-like) wrapper that uses pexpect to +interact with privacyIDEA's pam_python module for u2f challenge/response. + +Usage: + Make executable + Symlink ssh-u2f, scp-u2f, sftp-u2f, mosh-u2f, etc. into your PATH + Call just like ssh, eg. "ssh-u2f name@example.com" +""" + +ssh = None + +def handler(signum, frame): + global ssh + if ssh: + ssh.kill(signum) + sys.exit(signum) + +signal.signal(signal.SIGQUIT, handler) +signal.signal(signal.SIGTERM, handler) +signal.signal(signal.SIGINT, handler) + +def winch_handler(signum, frame): + global ssh + if ssh: + rows, cols = os.popen('stty size', 'r').read().split() + ssh.setwinsize(int(rows), int(cols)) + +signal.signal(signal.SIGWINCH, winch_handler) + +try: + command = os.path.splitext(os.path.basename(__file__))[0].split("-")[0] +except: + command = None + +ssh = pexpect.spawn(command or "ssh", sys.argv[1:]) +winch_handler(None, None) + +def passthrough(): + print() + sys.stdout.write(ssh.match.group()) + try: + ssh.interact() + except UnboundLocalError: + # Work around bug in pexpect 3.1 + pass + sys.exit(0) + +while True: + index = ssh.expect(["Authenticated with partial success.", + "([Pp]assword[^:\r\n]*|Enter additional factors): ?", + "----- BEGIN U2F CHALLENGE -----\r\n", + "[^ \r\n]+", + pexpect.EOF]) + + if index == 0: + print(ssh.match.group()) + + if index == 1: + try: + pin = getpass.getpass(ssh.match.group()) + except EOFError: + pin = "" + ssh.sendline(pin.strip()) + + elif index == 2: + u2f_origin = ssh.readline().strip() + u2f_challenge = ssh.readline().strip() + ssh.expect("(.*)----- END U2F CHALLENGE -----") + message = ssh.match.group(1).strip() + print(message or "Interact with your U2F token.") + p = subprocess.Popen(["u2f-host", "-aauthenticate", "-o", u2f_origin], + stdin=subprocess.PIPE, stdout=subprocess.PIPE) + out, err = p.communicate(u2f_challenge) + p.wait() + ssh.sendline(out.strip()) + + elif index == 3: + passthrough() + + elif index == 4: + sys.exit(0)