2
0
mirror of https://github.com/xcat2/confluent.git synced 2024-11-22 01:22:00 +00:00

Merge pull request #154 from tkucherera-lenovo/webauthn

Webauthn
This commit is contained in:
Jarrod Johnson 2024-09-12 13:25:15 -04:00 committed by GitHub
commit 942bbd1c69
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 378 additions and 81 deletions

View File

@ -1,110 +1,402 @@
import base64
from webauthn_rp.registrars import CredentialData
import confluent.tlvdata as tlvdata
import confluent.util as util
import json
import pywarp
import pywarp.backends
import pywarp.credentials
import secrets, time
from typing import Any, Optional
from webauthn_rp.backends import CredentialsBackend
from webauthn_rp.builders import *
from webauthn_rp.converters import cose_key, jsonify
from webauthn_rp.errors import WebAuthnRPError
from webauthn_rp.parsers import parse_cose_key, parse_public_key_credential
from webauthn_rp.registrars import *
from webauthn_rp.types import (
AttestationObject, AttestationType, AuthenticatorAssertionResponse,
AuthenticatorAttestationResponse, AuthenticatorData,
COSEAlgorithmIdentifier, PublicKeyCredential,
PublicKeyCredentialDescriptor, PublicKeyCredentialParameters,
PublicKeyCredentialRpEntity, PublicKeyCredentialType,
PublicKeyCredentialUserEntity, TrustedPath)
challenges = {}
class ConfluentBackend(pywarp.backends.CredentialStorageBackend):
def __init__(self, cfg):
self.cfg = cfg
CONFIG_MANAGER = None
def get_credential_ids_by_email(self, email):
if not isinstance(email, str):
email = email.decode('utf8')
authenticators = self.cfg.get_user(email).get('authenticators', {})
if not authenticators:
raise Exception('No authenticators found')
for cid in authenticators:
yield base64.b64decode(cid)
class Credential():
def __init__(self, id, signature_count, public_key):
self.id = id
self.signature_count = signature_count
self.credential_public_key = public_key
def get_credential_by_email_id(self, email, id):
if not isinstance(email, str):
email = email.decode('utf8')
authenticators = self.cfg.get_user(email).get('authenticators', {})
cid = base64.b64encode(id).decode('utf8')
pk = authenticators[cid]['cpk']
pk = base64.b64decode(pk)
return pywarp.credentials.Credential(credential_id=id, credential_public_key=pk)
class Challenge():
def __init__(self, request, timstamp_ms, id=None) -> None:
if id is None:
self.id = util.randomstring(16)
else:
self.id = id
self.request = request
self.timestamp_ms = timstamp_ms
def get_credential_by_email(self, email):
if not isinstance(email, str):
email = email.decode('utf8')
authenticators = self.cfg.get_user(email)
cid = list(authenticators)[0]
cred = authenticators[cid]
cid = base64.b64decode(cred['cid'])
cpk = base64.b64decode(cred['cpk'])
return pywarp.credentials.Credential(credential_id=cid, credential_public_key=cpk)
def save_credential_for_user(self, email, credential):
if not isinstance(email, str):
email = email.decode('utf8')
cid = base64.b64encode(credential.id).decode('utf8')
credential = {'cid': cid, 'cpk': base64.b64encode(bytes(credential.public_key)).decode('utf8')}
authenticators = self.cfg.get_user(email).get('authenticators', {})
authenticators[cid] = credential
self.cfg.set_user(email, {'authenticators': authenticators})
def save_challenge_for_user(self, email, challenge, type):
if not isinstance(email, str):
email = email.decode('utf8')
challenges[email] = challenge
class User():
def __init__(self, id, username, user_handle, challenge: Challenge = None, credential: Credential = None):
self.id = id
self.username = username
self.user_handle = user_handle
self.challenges = challenge
self.credentials = credential
def get_challenge_for_user(self, email, type):
if not isinstance(email, str):
email = email.decode('utf8')
return challenges[email]
def __parse_credentials(self):
return {"id": self.credentials.id, "signature_count": self.credentials.signature_count, "credential_public_key": self.credentials.credential_public_key}
def __parse_challenges(self):
return {"id": self.challenges.id, 'request': self.challenges.request, 'timestamp_ms': self.challenges.timestamp_ms}
@staticmethod
def seek_credential_by_id(credential_id):
"""
There certainly is a better way to do this but for now lets try the wrong way that works
"""
for username in CONFIG_MANAGER.list_users():
authenticators = CONFIG_MANAGER.get_user(username).get('authenticators', {})
try:
credential = authenticators['credentials']
except KeyError:
continue
if "id" in credential.keys() and credential["id"] == credential_id:
#for now leaving signature count as None
return (Credential(id=credential["id"], signature_count=None, public_key=credential["credential_public_key"]), username)
return None
@staticmethod
def get_credential(credential_id, username):
if not isinstance(username, str):
username = username.decode('utf8')
authenticators = CONFIG_MANAGER.get_user(username).get('authenticators', {})
try:
credential = authenticators['credentials']
except KeyError:
return None
if credential_id is None:
return Credential(id=credential["id"], signature_count=credential["signature_count"], public_key=credential["credential_public_key"])
if credential["id"] == credential_id:
return Credential(id=credential["id"], signature_count=credential["signature_count"], public_key=credential["credential_public_key"])
return None
@staticmethod
def get_challenge(challengeID, username):
if not isinstance(username, str):
username = username.decode('utf8')
authenticators = CONFIG_MANAGER.get_user(username).get('authenticators', {})
challenge = authenticators['challenges']
if challenge["id"] == challengeID:
return Challenge(request=challenge["request"], timstamp_ms=challenge["timestamp_ms"], id=challenge["id"])
return None
@staticmethod
def get(username):
if not CONFIG_MANAGER:
raise Exception('config manager is not set up')
if not isinstance(username, str):
username = username.decode('utf8')
userinfo = CONFIG_MANAGER.get_user(username)
authenticators = CONFIG_MANAGER.get_user(username).get('authenticators', {})
if userinfo is None:
return None
authid = userinfo.get('webauthid', None)
challenge = authenticators.get("challenges", None)
challenges_return = Challenge(challenge['request'], challenge['timestamp_ms'], id=challenge["id"])
credential = authenticators.get("credentials", None)
credentials_return = (Credential(credential['id'], credential['signature_count'], credential["credential_public_key"]))
return User(id=None, username=username, user_handle=authid, challenge=challenges_return, credential=credentials_return)
def save(self):
authenticators = CONFIG_MANAGER.get_user(self.username).get('authenticators', {})
authenticators['challenges'] = self.__parse_challenges() # Looks like the bigger the array we encounter problems changing to just save one challenge
authenticators['credentials'] = self.__parse_credentials()
CONFIG_MANAGER.set_user(self.username, {'authenticators': authenticators})
def add(self, item):
if isinstance(item, Challenge):
self.challenges = item
elif isinstance(item, Credential):
self.credentials = item
def update(self, item):
if isinstance(item, Challenge):
self.challenges = item
elif isinstance(item, Credential):
self.credentials = item
return
#raise Exception("Credential item not found")
def timestamp_ms():
return int(time.time() * 1000)
class RegistrarImpl(CredentialsRegistrar):
def register_credential_attestation(
self,
credential: PublicKeyCredential,
att: AttestationObject,
att_type: AttestationType,
user: PublicKeyCredentialUserEntity,
rp: PublicKeyCredentialRpEntity,
trusted_path: Optional[TrustedPath] = None) -> Any:
assert att.auth_data is not None
assert att.auth_data.attested_credential_data is not None
cpk = att.auth_data.attested_credential_data.credential_public_key
user_model = User.get(user.name)
if user_model is None:
return 'No user found'
credential_model = Credential(id=credential.raw_id, signature_count=None, public_key=cose_key(cpk))
user_model.add(credential_model)
user_model.save()
def register_credential_assertion(
self,
credential: PublicKeyCredential,
authenticator_data: AuthenticatorData,
user: PublicKeyCredentialUserEntity,
rp: PublicKeyCredentialRpEntity) -> Any:
user_model = User.get(user.name)
credential_model = User.get_credential(credential_id=credential.raw_id, username=user.name)
credential_model.signature_count = None
user_model.update(credential_model)
user_model.save()
def get_credential_data(
self,
credential_id: bytes) -> Optional[CredentialData]:
#credential_model = User.get_credential(credential_id=credential_id, username=username)
(credential_model, username) = User.seek_credential_by_id(credential_id)
user_model = User.get(username)
return CredentialData(
parse_cose_key(credential_model.credential_public_key),
credential_model.signature_count,
PublicKeyCredentialUserEntity(
name=user_model.username,
id=user_model.user_handle,
display_name=user_model.username
)
)
APP_ORIGIN = 'https://ndiamai'
APP_TIMEOUT = 60000
APP_RELYING_PARTY = PublicKeyCredentialRpEntity(name='Confluent Web UI', id="ndiamai")
APP_CCO_BUILDER = CredentialCreationOptionsBuilder(
rp=APP_RELYING_PARTY,
pub_key_cred_params=[
PublicKeyCredentialParameters(type=PublicKeyCredentialType.PUBLIC_KEY,
alg=COSEAlgorithmIdentifier.Value.ES256)
],
timeout=APP_TIMEOUT,
)
APP_CRO_BUILDER = CredentialRequestOptionsBuilder(
rp_id=APP_RELYING_PARTY.id,
timeout=APP_TIMEOUT,
)
APP_CREDENTIALS_BACKEND = CredentialsBackend(RegistrarImpl())
def registration_request(username, cfg):
user_model = User.get(username)
if user_model is None:
raise Exception("User not foud")
challenge_bytes = secrets.token_bytes(64)
challenge = Challenge(request=challenge_bytes, timstamp_ms=timestamp_ms())
user_model.add(challenge)
user_model.save()
options = APP_CCO_BUILDER.build(
user=PublicKeyCredentialUserEntity(
name=username,
id=user_model.user_handle,
display_name=username
),
challenge=challenge_bytes
)
options_json = jsonify(options)
return {
'challengeID': challenge.id,
'creationOptions': options_json
}
def registration_response(request, username):
try:
challengeID = request["challengeID"]
credential = parse_public_key_credential(json.loads(request["credential"]))
except Exception:
raise Exception("Could not parse input data")
if type(credential.response) is not AuthenticatorAttestationResponse:
raise Exception("Invalid response type")
challenge_model = User.get_challenge(challengeID, username)
if not challenge_model:
raise Exception("Could not find challenge matching given id")
user_model = User.get(username)
if not user_model:
raise Exception("Invalid Username")
current_timestamp = timestamp_ms()
if current_timestamp - challenge_model.timestamp_ms > APP_TIMEOUT:
return "Timeout"
user_entity = PublicKeyCredentialUserEntity(name=user_model.username, id=user_model.user_handle, display_name=user_model.username)
try:
APP_CREDENTIALS_BACKEND.handle_credential_attestation(
credential=credential,
user=user_entity,
rp=APP_RELYING_PARTY,
expected_challenge=challenge_model.request,
expected_origin=APP_ORIGIN
)
except WebAuthnRPError:
raise Exception("Could not handle credential attestation")
return True
def authentication_request(username):
user_model = User.get(username)
if user_model is None:
return 'User not registered'
credential = user_model.get_credential(None, username)
print(credential)
if credential is None:
return f'No credential for User found {username}'
challenge_bytes = secrets.token_bytes(64)
challenge = Challenge(request=challenge_bytes, timstamp_ms=timestamp_ms())
user_model.add(challenge)
user_model.save()
options = APP_CRO_BUILDER.build(
challenge=challenge_bytes,
allow_credentials=[
PublicKeyCredentialDescriptor(
id=credential.id,
type=PublicKeyCredentialType.PUBLIC_KEY
)
]
)
options_json = jsonify(options)
return {
'challengeID': challenge.id,
'requestOptions': options_json
}
def authentication_response(request, username):
try:
challengeID = request["challengeID"]
credential = parse_public_key_credential(json.loads(request["credential"]))
except Exception:
raise Exception("Could not parse input data")
if type(credential.response) is not AuthenticatorAssertionResponse:
raise Exception('Invalid response type')
challenge_model = User.get_challenge(challengeID, username)
if not challenge_model:
raise Exception("Could not find challenge matching given id")
user_model = User.get(username)
if not user_model:
raise Exception("Invalid Username")
current_timestamp = timestamp_ms()
if current_timestamp - challenge_model.timestamp_ms > APP_TIMEOUT:
return "Timeout"
user_entity = PublicKeyCredentialUserEntity(name=user_model.username, id=user_model.user_handle, display_name=user_model.username)
try:
APP_CREDENTIALS_BACKEND.handle_credential_assertion(
credential=credential,
user=user_entity,
rp=APP_RELYING_PARTY,
expected_challenge=challenge_model.request,
expected_origin=APP_ORIGIN
)
except WebAuthnRPError:
raise Exception('Could not handle credential assertion')
return {"verified": True}
def handle_api_request(url, env, start_response, username, cfm, headers, reqbody, authorized):
"""
For now webauth is going to be limited to just one passkey per user
If you try to register a new passkey this will just clear the old one and regist the new passkey
"""
global CONFIG_MANAGER
CONFIG_MANAGER = cfm
if env['REQUEST_METHOD'] != 'POST':
raise Exception('Only POST supported for webauthn operations')
url = url.replace('/sessions/current/webauthn', '')
if url == '/registration_options':
rp = pywarp.RelyingPartyManager('Confluent Web UI', credential_storage_backend=ConfluentBackend(cfm), require_attestation=False)
userinfo = cfm.get_user(username)
if not userinfo:
cfm.create_user(username, role='Stub')
userinfo = cfm.get_user(username)
authid = userinfo.get('authid', None)
authid = userinfo.get('webauthid', None)
if not authid:
authid = util.randomstring(64)
cfm.set_user(username, {'authid': authid})
opts = rp.get_registration_options(username)
# pywarp generates an id derived
# from username, which is a 'must not' in the spec
# we replace that with a complying approach
opts['user']['id'] = authid
if 'icon' in opts['user']:
del opts['user']['icon']
if 'id' in opts['rp']:
del opts['rp']['id']
authid = secrets.token_bytes(64)
cfm.set_user(username, {'webauthid': authid})
opts = registration_request(username, cfm)
start_response('200 OK', headers)
yield json.dumps(opts)
elif url.startswith('/registered_credentials/'):
username = url.rsplit('/', 1)[-1]
rp = pywarp.RelyingPartyManager('Confluent Web UI', credential_storage_backend=ConfluentBackend(cfm))
userinfo = cfm.get_user(username)
if not isinstance(username, bytes):
username = username.encode('utf8')
opts = rp.get_authentication_options(username)
opts['challenge'] = base64.b64encode(opts['challenge']).decode('utf8')
opts = authentication_request(username)
start_response('200 OK', headers)
yield json.dumps(opts)
elif url.startswith('/validate/'):
username = url.rsplit('/', 1)[-1]
userinfo = cfm.get_user(username)
if not isinstance(username, bytes):
username = username.encode('utf8')
rp = pywarp.RelyingPartyManager('Confluent Web UI', credential_storage_backend=ConfluentBackend(cfm))
req = json.loads(reqbody)
for x in req:
req[x] = base64.b64decode(req[x].replace('-', '+').replace('_', '/'))
req['email'] = username
rsp = rp.verify(**req)
if start_response:
rsp = authentication_response(req, username)
if rsp == 'Timeout':
start_response('408 Timeout', headers)
elif rsp['verified'] and start_response:
start_response('200 OK', headers)
sessinfo = {'username': username}
if 'authtoken' in authorized:
@ -116,13 +408,16 @@ def handle_api_request(url, env, start_response, username, cfm, headers, reqbody
else:
yield rsp
elif url == '/register_credential':
rp = pywarp.RelyingPartyManager('Confluent Web UI', credential_storage_backend=ConfluentBackend(cfm), require_attestation=False)
req = json.loads(reqbody)
for x in req:
req[x] = base64.b64decode(req[x].replace('-', '+').replace('_', '/'))
userinfo = cfm.get_user(username)
if not isinstance(username, bytes):
username = username.encode('utf8')
req['email'] = username
rsp = rp.register(**req)
start_response('200 OK', headers)
yield json.dumps(rsp)
rsp = registration_response(req, username)
if rsp == 'Timeout':
start_response('408 Timeout', headers)
else:
print('worked out')
start_response('200 OK', headers)
yield json.dumps({'status': 'Success'})

View File

@ -18,15 +18,17 @@ Prefix: %{_prefix}
BuildArch: noarch
Requires: confluent_vtbufferd
%if "%{dist}" == ".el7"
Requires: python-pyghmi >= 1.5.71, python-eventlet, python-greenlet, python-pycryptodomex >= 3.4.7, confluent_client == %{version}, python-pyparsing, python-paramiko, python-dnspython, python-netifaces, python2-pyasn1 >= 0.2.3, python-pysnmp >= 4.3.4, python-lxml, python-eficompressor, python-setuptools, python-dateutil, python-websocket-client python2-msgpack python-libarchive-c python-yaml python-monotonic cpio
Requires: python-pyghmi >= 1.0.34, python-eventlet, python-greenlet, python-pycryptodomex >= 3.4.7, confluent_client == %{version}, python-pyparsing, python-paramiko, python-webauthn-rp, python-dnspython, python-netifaces, python2-pyasn1 >= 0.2.3, python-pysnmp >= 4.3.4, python-lxml, python-eficompressor, python-setuptools, python-dateutil, python-websocket-client python2-msgpack python-libarchive-c python-yaml python-monotonic
%else
%if "%{dist}" == ".el8"
Requires: python3-pyghmi >= 1.5.71, python3-eventlet, python3-greenlet, python3-pycryptodomex >= 3.4.7, confluent_client == %{version}, python3-pyparsing, python3-paramiko, python3-dns, python3-netifaces, python3-pyasn1 >= 0.2.3, python3-pysnmp >= 4.3.4, python3-lxml, python3-eficompressor, python3-setuptools, python3-dateutil, python3-enum34, python3-asn1crypto, python3-cffi, python3-pyOpenSSL, python3-websocket-client python3-msgpack python3-libarchive-c python3-yaml openssl iproute cpio
Requires: python3-pyghmi >= 1.0.34, python3-eventlet, python3-greenlet, python3-pycryptodomex >= 3.4.7, confluent_client == %{version}, python3-pyparsing, python3-paramiko, python3-webauthn-rp, python3-dns, python3-netifaces, python3-pyasn1 >= 0.2.3, python3-pysnmp >= 4.3.4, python3-lxml, python3-eficompressor, python3-setuptools, python3-dateutil, python3-enum34, python3-asn1crypto, python3-cffi, python3-pyOpenSSL, python3-websocket-client python3-msgpack python3-libarchive-c python3-yaml openssl iproute
%else
%if "%{dist}" == ".el9"
Requires: python3-pyghmi >= 1.5.71, python3-eventlet, python3-greenlet, python3-pycryptodomex >= 3.4.7, confluent_client == %{version}, python3-pyparsing, python3-paramiko, python3-dns, python3-netifaces, python3-pyasn1 >= 0.2.3, python3-pysnmp >= 4.3.4, python3-lxml, python3-eficompressor, python3-setuptools, python3-dateutil, python3-cffi, python3-pyOpenSSL, python3-websocket-client python3-msgpack python3-libarchive-c python3-yaml openssl iproute cpio
Requires: python3-pyghmi >= 1.0.34, python3-eventlet, python3-greenlet, python3-pycryptodomex >= 3.4.7, confluent_client == %{version}, python3-pyparsing, python3-paramiko, python3-dns, python3-webauthn-rp, python3-netifaces, python3-pyasn1 >= 0.2.3, python3-pysnmp >= 4.3.4, python3-lxml, python3-eficompressor, python3-setuptools, python3-dateutil, python3-cffi, python3-pyOpenSSL, python3-websocket-client python3-msgpack python3-libarchive-c python3-yaml openssl iproute
%else
Requires: python3-dbm,python3-pyghmi >= 1.5.71, python3-eventlet, python3-greenlet, python3-pycryptodome >= 3.4.7, confluent_client == %{version}, python3-pyparsing, python3-paramiko, python3-dnspython, python3-netifaces, python3-pyasn1 >= 0.2.3, python3-pysnmp >= 4.3.4, python3-lxml, python3-eficompressor, python3-setuptools, python3-dateutil, python3-cffi, python3-pyOpenSSL, python3-websocket-client python3-msgpack python3-libarchive-c python3-PyYAML openssl iproute
Requires: python3-dbm,python3-pyghmi >= 1.0.34, python3-eventlet, python3-greenlet, python3-pycryptodome >= 3.4.7, confluent_client == %{version}, python3-pyparsing, python3-paramiko, python3-webauthn-rp, python3-dnspython, python3-netifaces, python3-pyasn1 >= 0.2.3, python3-pysnmp >= 4.3.4, python3-lxml, python3-eficompressor, python3-setuptools, python3-dateutil, python3-cffi, python3-pyOpenSSL, python3-websocket-client python3-msgpack python3-libarchive-c python3-PyYAML openssl iproute
%endif
%endif
%endif