From f858a1a78c7d1e8d516005ffc9aa5b44273ac07e Mon Sep 17 00:00:00 2001 From: tuxcoder Date: Mon, 25 Dec 2023 17:28:09 +0100 Subject: [PATCH] add basic passkey management --- .../versions/b5448df204eb_passkey.py | 40 ++++ lenticular_cloud/model.py | 18 +- lenticular_cloud/template/base.html.j2 | 4 +- .../template/frontend/base.html.j2 | 3 +- .../template/frontend/passkey_list.html.j2 | 30 +++ .../template/frontend/passkey_new.html.j2 | 69 +++++++ .../template/frontend/webauthn_list.html | 30 --- .../template/frontend/webauthn_register.html | 140 ------------- lenticular_cloud/views/frontend.py | 193 ++++++++++-------- overlay.nix | 4 +- 10 files changed, 258 insertions(+), 273 deletions(-) create mode 100644 lenticular_cloud/migrations/versions/b5448df204eb_passkey.py create mode 100644 lenticular_cloud/template/frontend/passkey_list.html.j2 create mode 100644 lenticular_cloud/template/frontend/passkey_new.html.j2 delete mode 100644 lenticular_cloud/template/frontend/webauthn_list.html delete mode 100644 lenticular_cloud/template/frontend/webauthn_register.html diff --git a/lenticular_cloud/migrations/versions/b5448df204eb_passkey.py b/lenticular_cloud/migrations/versions/b5448df204eb_passkey.py new file mode 100644 index 0000000..e7ff0ad --- /dev/null +++ b/lenticular_cloud/migrations/versions/b5448df204eb_passkey.py @@ -0,0 +1,40 @@ +"""passkey + +Revision ID: b5448df204eb +Revises: a74320a5d7a1 +Create Date: 2023-12-25 00:13:01.703575 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'b5448df204eb' +down_revision = 'a74320a5d7a1' +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('passkey_credential', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('user_id', sa.Uuid(), nullable=False), + sa.Column('credential_id', sa.BINARY(), nullable=False), + sa.Column('credential_public_key', sa.BINARY(), nullable=False), + sa.Column('name', sa.String(length=250), nullable=False), + sa.Column('last_used', sa.DateTime(), nullable=True), + sa.Column('sign_count', sa.Integer(), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.Column('modified_at', sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(['user_id'], ['user.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id') + ) + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table('passkey_credential') + # ### end Alembic commands ### diff --git a/lenticular_cloud/model.py b/lenticular_cloud/model.py index acfbf0f..9da828f 100644 --- a/lenticular_cloud/model.py +++ b/lenticular_cloud/model.py @@ -164,14 +164,11 @@ class User(BaseModel, ModelUpdatedMixin): app_tokens: Mapped[List['AppToken']] = relationship('AppToken', back_populates='user') # totps: Mapped[List['Totp']] = relationship('Totp', back_populates='user', default_factory=list) - # webauthn_credentials: Mapped[List['WebauthnCredential']] = relationship('WebauthnCredential', back_populates='user', cascade='delete,delete-orphan', passive_deletes=True, default_factory=list) + passkey_credentials: Mapped[List['PasskeyCredential']] = relationship('PasskeyCredential', back_populates='user', cascade='delete,delete-orphan', passive_deletes=True) @property def totps(self) -> List['Totp']: return [] - @property - def webauthn_credentials(self) -> List['WebauthnCredential']: - return [] def __init__(self, **kwargs) -> None: super().__init__(**kwargs) @@ -245,17 +242,18 @@ class Totp(BaseModel, ModelUpdatedMixin): return totp.verify(token) -class WebauthnCredential(BaseModel, ModelUpdatedMixin): # pylint: disable=too-few-public-methods - """Webauthn credential model""" +class PasskeyCredential(BaseModel, ModelUpdatedMixin): # pylint: disable=too-few-public-methods + """Passkey credential model""" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) user_id: Mapped[uuid.UUID] = mapped_column(db.Uuid, db.ForeignKey('user.id', ondelete='CASCADE'), nullable=False) - user_handle: Mapped[str] = mapped_column(db.String(64), nullable=False) - credential_data: Mapped[bytes] = mapped_column(db.LargeBinary, nullable=False) + credential_id: Mapped[bytes] = mapped_column(db.BINARY, nullable=False) + credential_public_key: Mapped[bytes] = mapped_column(db.BINARY, nullable=False) name: Mapped[str] = mapped_column(db.String(250), nullable=False) - registered: Mapped[datetime] = mapped_column(db.DateTime, default=datetime.utcnow, nullable=False) + last_used: Mapped[Optional[datetime]] = mapped_column(db.DateTime, nullable=True, default=None) + sign_count: Mapped[int] = mapped_column(db.Integer, nullable=False, default=0) - # user = db.relationship('User', back_populates='webauthn_credentials') + user = db.relationship('User', back_populates='passkey_credentials') class Group(BaseModel, ModelUpdatedMixin): diff --git a/lenticular_cloud/template/base.html.j2 b/lenticular_cloud/template/base.html.j2 index b507d82..9277d59 100644 --- a/lenticular_cloud/template/base.html.j2 +++ b/lenticular_cloud/template/base.html.j2 @@ -213,9 +213,9 @@ action_text - text of submit button class_ - sets a class for form #} -{% macro render_form(form, action_url='', action_text='Submit', class_='', btn_class='btn btn-primary', method='post', onsubmit='') -%} +{% macro render_form(form, action_url='', id='', action_text='Submit', class_='', btn_class='btn btn-primary', method='post', onsubmit='') -%} -
+ {{ _render_form(form) }} {% if not form.submit %} diff --git a/lenticular_cloud/template/frontend/base.html.j2 b/lenticular_cloud/template/frontend/base.html.j2 index 670613e..42ff924 100644 --- a/lenticular_cloud/template/frontend/base.html.j2 +++ b/lenticular_cloud/template/frontend/base.html.j2 @@ -17,7 +17,8 @@
{#logo#}
  • {{ gettext('Account') }}
    • -
  • {{ gettext('Client Cert') }}
    • + {#
  • {{ gettext('Client Cert') }}
    • #} +
  • {{ gettext('Passkey') }}
  • {{ gettext('App Tokens') }}
  • {{ gettext('2FA - TOTP') }}
  • {{ gettext('Oauth2 Tokens') }}
    • diff --git a/lenticular_cloud/template/frontend/passkey_list.html.j2 b/lenticular_cloud/template/frontend/passkey_list.html.j2 new file mode 100644 index 0000000..f8bf437 --- /dev/null +++ b/lenticular_cloud/template/frontend/passkey_list.html.j2 @@ -0,0 +1,30 @@ +{% extends 'frontend/base.html.j2' %} + +{% block content %} +
    +

    Passkey Credentials list

    + + + + + + + + + + + + + {% for credential in credentials %} + + + + + + + + {% endfor %} + +
    nameidlast usedcreated atactions
    {{ credential.name }}{{ credential.credential_id[0:8].hex() }}...{{ credential.last_used }}{{ credential.created_at }}...{{ render_form(button_form, action_url=url_for('.passkey_delete', id=credential.id), action_text='delete', btn_class='btn btn-danger') }}
    +
    +{% endblock %} diff --git a/lenticular_cloud/template/frontend/passkey_new.html.j2 b/lenticular_cloud/template/frontend/passkey_new.html.j2 new file mode 100644 index 0000000..04a3dbb --- /dev/null +++ b/lenticular_cloud/template/frontend/passkey_new.html.j2 @@ -0,0 +1,69 @@ +{#- This file is part of sner4 project governed by MIT license, see the LICENSE.txt file. -#} +{% extends 'frontend/base.html.j2' %} + +{% block script %} + +{% endblock %} + +{% block content %} +
    +

    Register new Passkey credential

    + +
    + To register new credential: +
      +
    1. Insert/connect authenticator and verify user presence.
      2. +
    2. Set name for the new credential.
      3. +
    3. Submit the registration.
      4. +
    +
    + + {{ render_form(form, id="webauthn_register_form") }} +
    +{% endblock %} diff --git a/lenticular_cloud/template/frontend/webauthn_list.html b/lenticular_cloud/template/frontend/webauthn_list.html deleted file mode 100644 index 633eab0..0000000 --- a/lenticular_cloud/template/frontend/webauthn_list.html +++ /dev/null @@ -1,30 +0,0 @@ -{% extends 'frontend/base.html.j2' %} - -{% block content %} -
    -

    WebauthnCredentials list

    - - - - - - - - - - - - - {% for cred in creds %} - - - - - - - - {% endfor %} - -
    user.usernameuser_handlecredential_dataname_actions
    {{ cred.user.username }}{{ cred.user_handle }}{{ cred.credential_data[0:40] }}...{{ cred.name }}{{ render_form(button_form, action_url=url_for('app.webauthn_delete_route', webauthn_id=cred.id)) }}
    -
    -{% endblock %} diff --git a/lenticular_cloud/template/frontend/webauthn_register.html b/lenticular_cloud/template/frontend/webauthn_register.html deleted file mode 100644 index b6f7a27..0000000 --- a/lenticular_cloud/template/frontend/webauthn_register.html +++ /dev/null @@ -1,140 +0,0 @@ -{#- This file is part of sner4 project governed by MIT license, see the LICENSE.txt file. -#} -{% extends 'frontend/base.html.j2' %} - -{% block script %} - -{% endblock %} - -{% block content %} -
    -

    Register new Webauthn credential

    - -
    - To register new credential: -
      -
    1. Insert/connect authenticator and verify user presence.
      2. -
    2. Optionaly set comment for the new credential.
      3. -
    3. Submit the registration.
      4. -
    -
    - - {{ render_form(form) }} - {# - - {{ form.csrf_token }} -
    - -

    To be prepared

    -
    - {{ b_wtf.bootstrap_field(form.attestation, horizontal=True) }} - {{ b_wtf.bootstrap_field(form.name, horizontal=True) }} - {{ b_wtf.bootstrap_field(form.submit, horizontal=True) }} - - #} -
    -{% endblock %} diff --git a/lenticular_cloud/views/frontend.py b/lenticular_cloud/views/frontend.py index b1827cf..edc7034 100644 --- a/lenticular_cloud/views/frontend.py +++ b/lenticular_cloud/views/frontend.py @@ -1,29 +1,24 @@ from authlib.integrations.base_client.errors import MissingTokenError, InvalidTokenError from base64 import b64encode, b64decode -from fido2 import cbor -from fido2.webauthn import CollectedClientData, AttestationObject, AttestedCredentialData, AuthenticatorData, PublicKeyCredentialUserEntity -from flask import Blueprint, Response, redirect, request +from flask import Blueprint, redirect, request from flask import current_app -from flask import jsonify, session, flash +from flask import jsonify, session from flask import render_template, url_for -from flask_login import login_user, logout_user, current_user +from flask_login import logout_user, current_user from http import HTTPStatus from werkzeug.utils import redirect import logging from datetime import timedelta -from base64 import b64decode from flask.typing import ResponseReturnValue -from oauthlib.oauth2.rfc6749.errors import TokenExpiredError from ory_hydra_client.api.o_auth_2 import list_o_auth_2_consent_sessions, revoke_o_auth_2_consent_sessions from ory_hydra_client.models import GenericError -from urllib.parse import urlencode, parse_qs -from random import SystemRandom -import string -from collections.abc import Iterable -from typing import Optional, Mapping, Iterator, List, Any +from urllib.parse import urlparse +from typing import Optional, Any +import jwt +from datetime import datetime, timedelta -from ..model import db, User, SecurityUser, Totp, AppToken, WebauthnCredential +from ..model import db, User, Totp, AppToken, PasskeyCredential from ..form.frontend import ClientCertForm, TOTPForm, \ TOTPDeleteForm, PasswordChangeForm, WebauthnRegisterForm, \ AppTokenForm, AppTokenDeleteForm @@ -223,96 +218,118 @@ def totp_delete(totp_name) -> ResponseReturnValue: 'status': 'ok'}) -@frontend_views.route('/webauthn/list', methods=['GET']) -def webauthn_list_route() -> ResponseReturnValue: + +## Passkey + +import webauthn +from webauthn.helpers.structs import ( + AuthenticatorSelectionCriteria, + PublicKeyCredentialDescriptor, + ResidentKeyRequirement, + UserVerificationRequirement, +) + +@frontend_views.route('/passkey/list', methods=['GET']) +def passkey() -> ResponseReturnValue: """list registered credentials for current user""" - creds = WebauthnCredential.query.all() # type: Iterable[WebauthnCredential] - return render_template('frontend/webauthn_list.html', creds=creds, button_form=ButtonForm()) + credentials = PasskeyCredential.query.all() + return render_template('frontend/passkey_list.html.j2', credentials=credentials, button_form=ButtonForm()) -@frontend_views.route('/webauthn/delete/', methods=['POST']) -def webauthn_delete_route(webauthn_id: str) -> ResponseReturnValue: + +@frontend_views.route('/passkey/new', methods=['GET']) +def passkey_new() -> ResponseReturnValue: + """register credential for current user""" + public_url = urlparse(current_app.config['PUBLIC_URL']) + + + user = get_current_user() # type: User + form = WebauthnRegisterForm() + + options = webauthn.generate_registration_options( + rp_name="Lenticluar Cloud", + rp_id=public_url.hostname, + user_id=str(user.id), + user_name=user.username, + authenticator_selection=AuthenticatorSelectionCriteria( + user_verification=UserVerificationRequirement.REQUIRED, + resident_key=ResidentKeyRequirement.REQUIRED, + ), + exclude_credentials = list(map(lambda x: PublicKeyCredentialDescriptor(id=x.credential_id), user.passkey_credentials)) + ) + secret_key = current_app.config['SECRET_KEY'] + token = jwt.encode({ + 'challenge': b64encode(options.challenge).decode(), + 'iat': datetime.utcnow() - timedelta(minutes=1), + 'nbf': datetime.utcnow(), + 'exp': datetime.utcnow() + timedelta(minutes=15), + }, secret_key, algorithm="HS256" + ) + + return render_template( + 'frontend/passkey_new.html.j2', + form=form, + options=webauthn.options_to_json(options), + token=token, + ) + +@frontend_views.route('/passkey/new', methods=['POST']) +def passkey_new_process() -> ResponseReturnValue: + secret_key = current_app.config['SECRET_KEY'] + public_url = urlparse(current_app.config['PUBLIC_URL']) + user = get_current_user() + + data = request.get_json() + try: + token = jwt.decode( + data['token'], secret_key, algorithms=['HS256'], + options = { + 'require': ["challenge", "exp", "iat", "nbf"], + + }) + except jwt.exceptions.MissingRequiredClaimError: + return jsonify({'message': "invalid token"}), 400 + challenge = b64decode(token['challenge']) + credential = data['credential'] + name = data['name'] + + result = webauthn.verify_registration_response( + credential = credential, + expected_rp_id = public_url.hostname, + expected_challenge = challenge, + expected_origin = [ public_url.geturl() ], + ) + if not result.user_verified: + return jsonify({ "message": "invalid auth" }), 403 + + db.session.add(PasskeyCredential( + id=None, + user_id=user.id, + credential_id=result.credential_id, + credential_public_key=result.credential_public_key, + name=name, + )) + db.session.commit() + logger.info(f"add new passkey for user {user.username}") + + return jsonify({}) + +@frontend_views.route('/passkey/delete/', methods=['POST']) +def passkey_delete(id: str) -> ResponseReturnValue: """delete registered credential""" form = ButtonForm() if form.validate_on_submit(): - cred = WebauthnCredential.query.filter(WebauthnCredential.id == webauthn_id).one() # type: WebauthnCredential + cred = PasskeyCredential.query.filter(PasskeyCredential.id == id).first_or_404() db.session.delete(cred) db.session.commit() - return redirect(url_for('app.webauthn_list_route')) + return redirect(url_for('.passkey')) return '', HTTPStatus.BAD_REQUEST - -def webauthn_credentials(user: User) -> list[AttestedCredentialData]: - """get and decode all credentials for given user""" - - def decode(creds: List[WebauthnCredential]) -> Iterator[AttestedCredentialData]: - for cred in creds: - data = cbor.decode(cred.credential_data) - if isinstance(data, Mapping): - yield AttestedCredentialData.create(**data) - - return list(decode(user.webauthn_credentials)) - - -def random_string(length=32) -> str: - """generates random string""" - return ''.join([SystemRandom().choice(string.ascii_letters + string.digits) for i in range(length)]) - - -@frontend_views.route('/webauthn/pkcco', methods=['POST']) -def webauthn_pkcco_route() -> ResponseReturnValue: - """get publicKeyCredentialCreationOptions""" - - user = User.query.get(get_current_user().id) #type: Optional[User] - if user is None: - return 'internal error', 500 - user_handle = random_string() - exclude_credentials = webauthn_credentials(user) - pkcco, state = webauthn.register_begin( - user=PublicKeyCredentialUserEntity(id=user_handle.encode('utf-8'), name=user.username, display_name=user.username), - credentials=exclude_credentials - ) - session['webauthn_register_user_handle'] = user_handle - session['webauthn_register_state'] = state - return Response(b64encode(cbor.encode(pkcco)).decode('utf-8'), mimetype='text/plain') - - -@frontend_views.route('/webauthn/register', methods=['GET', 'POST']) -def webauthn_register_route() -> ResponseReturnValue: - """register credential for current user""" - - user = get_current_user() # type: User - form = WebauthnRegisterForm() - if form.validate_on_submit(): - try: - attestation = cbor.decode(b64decode(form.attestation.data)) - if not isinstance(attestation, Mapping) or 'clientDataJSON' not in attestation or 'attestationObject' not in attestation: - return 'invalid attestion data', 400 - auth_data = webauthn.register_complete( - session.pop('webauthn_register_state'), - CollectedClientData(attestation['clientDataJSON']), - AttestationObject(attestation['attestationObject'])) - - db.session.add(WebauthnCredential( - user=user, - user_handle=session.pop('webauthn_register_user_handle'), - credential_data=cbor.encode(auth_data.credential_data.__dict__), - name=form.name.data)) - db.session.commit() - - return redirect(url_for('app.webauthn_list_route')) - except (KeyError, ValueError) as e: - logger.exception(e) - flash('Error during registration.', 'error') - - return render_template('frontend/webauthn_register.html', form=form) - - @frontend_views.route('/password_change') def password_change() -> ResponseReturnValue: form = PasswordChangeForm() diff --git a/overlay.nix b/overlay.nix index 51f8ed3..c42e23b 100644 --- a/overlay.nix +++ b/overlay.nix @@ -86,12 +86,12 @@ in { cryptography blinker authlib # as oauth client lib - fido2 # for webauthn flask_migrate # db migrations flask-dance ory-hydra-client toml - webauthn + webauthn pyopenssl + pyjwt pkgs.nodejs #node-env