diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index f6a0cba6fa4b0c27a1a986d58bff05b88f249f5c..f9a94ced3017eb3e4a4af528e4c35426ad599d13 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -228,4 +228,6 @@ pylint-lint:
PYLINT_PLUGINS: "pylint_flask pylint_flask_sqlalchemy"
image: ${CI_REGISTRY_IMAGE}/pylint:${CI_COMMIT_REF_NAME}
script:
- - darker --check -i -L pylint --diff --revision remotes/origin/main .
+ # Run darker with --diff command. This will throw exit code 1 if there are
+ # lint errors, but a 0 if there are only formatting recommendations
+ - darker -i -L pylint --diff --revision remotes/origin/main .
diff --git a/login/app.py b/login/app.py
index 68f45284d6b0329f74c8fd89af3bdab73d145396..7491b9b63b242f40e26c053d848a6973a0026426 100644
--- a/login/app.py
+++ b/login/app.py
@@ -1,16 +1,23 @@
+"""Flask application which provides the interface of a login panel. The
+application interacts with different backend, like the Kratos backend for users,
+Hydra for OIDC sessions and Postgres for application and role specifications.
+The application provides also several command line options to interact with
+the user entries in the database(s)"""
+
# Basic system imports
-import pprint
import logging
import os
-import click
import urllib.parse
+import urllib.request
+import click
# Flask
from flask import abort, Flask, redirect, request, render_template
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
+# False positive: pylint: disable=ungrouped-imports
from flask.cli import AppGroup
@@ -21,20 +28,15 @@ import hydra_client
# Kratos, Identity manager
import ory_kratos_client
-from ory_kratos_client.api import metadata_api
from ory_kratos_client.api import v0alpha2_api as kratos_api
-from ory_kratos_client.model.generic_error import GenericError
-from ory_kratos_client.model.inline_response200 import InlineResponse200
-from ory_kratos_client.model.inline_response2001 import InlineResponse2001
-from ory_kratos_client.model.inline_response503 import InlineResponse503
-from ory_kratos_client.model.admin_create_identity_body import AdminCreateIdentityBody
-from ory_kratos_client.model.admin_create_self_service_recovery_link_body import AdminCreateSelfServiceRecoveryLinkBody
-from ory_kratos_client.model.identity_state import IdentityState
+from exceptions import BackendError
+from kratos import KratosUser
-# Initaliaze the FLASK app
+
+# Initialize the FLASK app
app = Flask(__name__,
static_url_path='/static')
@@ -78,17 +80,19 @@ KRATOS_PUBLIC = kratos_api.V0alpha2Api(ory_kratos_client.ApiClient(tmp))
# Create DB & migrate interface
db = SQLAlchemy(app)
-migrate = Migrate(app, db)
+Migrate(app, db)
-# Import models
+# Import needs to happen after the database interface is created. AppRole is
+# imported for future usage
+# pylint: disable=wrong-import-position unused-import
from models import User, App, AppRole
#
# WARNING:
#
# Below are very minimalistic calls to interfaces for development and testing
-# purposed. Eventually this need to be moved to seperate files and more
-# sophisticated calls with try{} catch{} claused etc.
+# purposed. Eventually this need to be moved to separate files and more
+# sophisticated calls with try{} catch{} clauses etc.
#
##############################################################################
@@ -104,6 +108,10 @@ app_cli = AppGroup('app')
@click.argument('slug')
@click.argument('name')
def create_app(slug, name):
+ """Adds an app into the database
+ :param slug: str short name of the app
+ :param name: str name of the application
+ """
app.logger.info(f"Creating app definition: {name} ({slug})")
obj = App()
@@ -117,17 +125,21 @@ def create_app(slug, name):
@app_cli.command('list')
def list_app():
+ """List all apps found in the database"""
app.logger.info("Listing configured apps")
apps = App.query.all()
for obj in apps:
- print("App name: %s \t Slug: %s" %(obj.name, obj.slug))
+ print(f"App name: {obj.name} \t Slug: {obj.slug}")
@app_cli.command('delete',)
@click.argument('slug')
def delete_app(slug):
- app.logger.info("Trying to delete app: {0}".format(slug))
+ """Removes app from database
+ :param slug: str Slug of app to remove
+ """
+ app.logger.info(f"Trying to delete app: {slug}")
obj = App.query.filter_by(slug=slug).first()
if not obj:
@@ -135,8 +147,9 @@ def delete_app(slug):
return
- # TODO: Deleting will (propably) fail is there are still roles attached
- # these probobly need to be cleaned up first.
+ # Deleting will (probably) fail is there are still roles attached. This is a
+ # PoC implementation only. Actually management of apps and roles will be
+ # done by the backend application
db.session.delete(obj)
db.session.commit()
app.logger.info("Success")
@@ -147,146 +160,153 @@ app.cli.add_command(app_cli)
## CLI USER COMMANDS
-
-@user_cli.command('create')
+@user_cli.command('show')
@click.argument('email')
-def create_user(email):
- app.logger.info("Creating user with email: ({0})".format(email))
-
- obj = User()
- obj.email = email
-
-
- # TODO:
- # - Should check if database entry already exists. If so, double check if
- # kratos user exists. If not, create that one.
- #
- # - If no DB user, should check with kratos of user already exists. If so,
- # throw warning, but still create DB entrie
- #
- # - After creating kratos user, check if success, otherwise throw warning.
-
- body = AdminCreateIdentityBody(
- schema_id="default",
- traits={'email':email},
- ) # AdminCreateIdentityBody | (optional)
-
- #state=IdentityState("active"),
-
- kratos_obj = KRATOS_ADMIN.admin_create_identity(admin_create_identity_body=body)
+def show_user(email):
+ """Show user details. Output a table with the user and details about the
+ internal state/values of the user object
+ :param email: Email address of the user to show
+ """
+ user = KratosUser.find_by_email(KRATOS_ADMIN, email)
+ print(user)
+ print("")
+ print(f"UUID: {user.uuid}")
+ print(f"Updated: {user.updated_at}")
+ print(f"Created: {user.created_at}")
+ print(f"State: {user.state}")
+
+@user_cli.command('update')
+@click.argument('email')
+@click.argument('field')
+@click.argument('value')
+def update_user(email, field, value):
+ """Update an user object. It can modify email and name currently
+ :param email: Email address of user to update
+ :param field: Field to update, supported [name|email]
+ :param value: The value to set the field with
+ """
+ app.logger.info(f"Looking for user with email: {email}")
+ user = KratosUser.find_by_email(KRATOS_ADMIN, email)
+ if not user:
+ app.logger.error(f"User with email {email} not found.")
+ return
+ if field == 'name':
+ user.name = value
+ elif field == 'email':
+ user.email = value
+ else:
+ app.logger.error(f"Field not found: {field}")
- obj.kratos_id = kratos_obj.id
+ user.save()
- db.session.add(obj)
-
- db.session.commit()
+@user_cli.command('delete')
+@click.argument('email')
+def delete_user(email):
+ """Delete an user from the database
+ :param email: Email address of user to delete
+ """
+ app.logger.info(f"Looking for user with email: {email}")
+ user = KratosUser.find_by_email(KRATOS_ADMIN, email)
+ if not user:
+ app.logger.error(f"User with email {email} not found.")
+ return
+ user.delete()
-@user_cli.command('list')
-def list_user():
- app.logger.info("Listing users")
- users = User.query.all()
- for obj in users:
- print("Email: %s (admin: %s)" %(obj.email, obj.admin))
+@user_cli.command('create')
+@click.argument('email')
+def create_user(email):
+ """Create a user in the kratos database. The argument must be an unique
+ email address
+ :param email: string Email address of user to add
+ """
+ app.logger.info(f"Creating user with email: ({email})")
+ # Create a user
+ user = KratosUser(KRATOS_ADMIN)
+ user.email = email
+ user.save()
-@user_cli.command('delete',)
+@user_cli.command('setpassword')
@click.argument('email')
-def delete_user(email):
- app.logger.info("Trying to delete user: {0}".format(email))
- obj = User.query.filter_by(email=email).first()
+@click.argument('password')
+def setpassword_user(email, password):
+ """Set a password for an account
+ :param email: email address of account to set a password for
+ :param password: password to be set
+ :return: true on success, false if not set (too weak)
+ :rtype: boolean
+ :raise: exception if unexepted error happens
+ """
- if not obj:
- app.logger.info("Not found")
- return
+ app.logger.info(f"Creating user with email: ({email})")
- db.session.delete(obj)
+ # Kratos does not provide an interface to set a password directly. However
+ # we still want to be able to set a password. So we have to hack our way
+ # a bit around this. We do this by creating a recovery link though the
+ # admin interface (which is not e-mailed) and then follow the recovery
+ # flow in the public facing pages of kratos
- # TODO:
- # - if delete succesfull, also delete kratos user, if exists
- # - probably user roles need to be deleted before user can be deleted
+ try:
+ # Get the ID of the user
+ kratos_user = KratosUser.find_by_email(KRATOS_ADMIN, email)
- db.session.commit()
- app.logger.info("Success")
- return
+ # Get a recovery URL
+ url = kratos_user.get_recovery_link()
-@user_cli.command('setadmin')
-@click.argument('email')
-def setadmin_user(email):
- app.logger.info("Trying to make user into admin: {0}".format(email))
- obj = User.query.filter_by(email=email).first()
+ # Execute UI sequence to set password, given we have a recovery URL
+ result = kratos_user.ui_set_password(app.config["KRATOS_PUBLIC_URL"], url, password)
- if not obj:
- app.logger.info("Not found")
- return
+ except BackendError as error:
+ app.logger.error(f"Error while setting password: {error}")
+ return False
- obj.admin = True
- db.session.commit()
- app.logger.info("Success")
- return
+ if result:
+ app.logger.info("Success setting password")
+ else:
+ app.logger.error("Failed to set password. Password too weak?")
+ return result
-@user_cli.command('unsetadmin')
-@click.argument('email')
-def unsetadmin_user(email):
- app.logger.info("Trying to make user into normal user: {0}".format(email))
- obj = User.query.filter_by(email=email).first()
- if not obj:
- app.logger.info("Not found")
- return
- obj.admin = False
- db.session.commit()
- app.logger.info("Success")
- return
+@user_cli.command('list')
+def list_user():
+ """Show a list of users in the database"""
+ app.logger.info("Listing users")
+ users = KratosUser.find_all(KRATOS_ADMIN)
+ for obj in users:
+ print(obj)
@user_cli.command('recover')
@click.argument('email')
def recover_user(email):
- app.logger.info("Trying to send recover email for user: {0}".format(email))
- obj = User.query.filter_by(email=email).first()
-
-
- if not obj:
- app.logger.info("Not found")
- return
-
- if not obj.kratos_id:
- app.logger.info("User found, but no kratos ID")
- return
+ """Get recovery link for a user, to manual update the user/use
+ :param email: Email address of the user
+ """
- body = AdminCreateSelfServiceRecoveryLinkBody(
- expires_in="1h",
- identity_id=obj.kratos_id,
- )
+ app.logger.info(f"Trying to send recover email for user: {email}")
- # example passing only required values which don't have defaults set
- # and optional values
try:
- # Create a Recovery Link
- api_response = KRATOS_ADMIN.admin_create_self_service_recovery_link(
- admin_create_self_service_recovery_link_body=body)
+ # Get the ID of the user
+ kratos_user = KratosUser.find_by_email(KRATOS_ADMIN, email)
- pprint.pprint(api_response)
- except ory_kratos_client.ApiException as error:
- app.logger.error("Exception when calling" +
- "V0alpha2Api->admin_create_self_service_recovery_link: %s\n" % error)
+ # Get a recovery URL
+ url = kratos_user.get_recovery_link()
-
- return
+ print(url)
+ except BackendError as error:
+ app.logger.error(f"Error while getting reset link: {error}")
app.cli.add_command(user_cli)
-
-
##############################################################################
# WEB ROUTES #
##############################################################################
@@ -332,7 +352,7 @@ def settings():
@app.route('/login', methods=['GET', 'POST'])
def login():
"""Start login flow
- If already logged in, shows the loggedin template. Otherwise creates a login
+ If already logged in, shows the loggedin template. Otherwise creates a login
flow, if no active flow will redirect to kratos to create a flow.
:param flow: flow as given by Kratos
@@ -340,7 +360,7 @@ def login():
"""
# Check if we are logged in:
- profile = getAuth()
+ profile = get_auth()
if profile:
return render_template(
@@ -350,7 +370,7 @@ def login():
flow = request.args.get("flow")
- # If we do not have a flow, get one.
+ # If we do not have a flow, get one.
if not flow:
return redirect(app.config["KRATOS_PUBLIC_URL"] + "self-service/login/browser")
@@ -364,7 +384,7 @@ def login():
def auth():
"""Authorize an user for an application
If an application authenticated against the IdP (Idenitity Provider), if
- there are no active session, the user is forwarded to the login page.
+ there are no active session, the user is forwarded to the login page.
This is the entry point for those authorization requests. The challenge
as provided, is verified. If an active user is logged in, the request
is accepted and the user is returned to the application. If the user is not
@@ -375,8 +395,8 @@ def auth():
challenge = None
- # Retrieve the challenge id from the request. Depending on the method it is
- # saved in the form (POST) or in a GET variable. If this variable is not set
+ # Retrieve the challenge id from the request. Depending on the method it is
+ # saved in the form (POST) or in a GET variable. If this variable is not set
# we can not continue.
if request.method == 'GET':
challenge = request.args.get("login_challenge")
@@ -384,13 +404,12 @@ def auth():
challenge = request.args.post("login_challenge")
if not challenge:
- # TODO: Use local error page?
- app.logger.error("No challange given. Error in request")
- abort(400)
+ app.logger.error("No challenge given. Error in request")
+ abort(400, description="Challenge required when requesting authorization")
# Check if we are logged in:
- profile = getAuth()
+ profile = get_auth()
# If the user is not logged in yet, we redirect to the login page
@@ -399,7 +418,7 @@ def auth():
# The redirect URL is back to this page (auth) with the same challenge
# so we can pickup the flow where we left off.
if not profile:
- url = app.config["PUBLIC_URL"] + "/auth?login_challenge=" + challenge;
+ url = app.config["PUBLIC_URL"] + "/auth?login_challenge=" + challenge
url = urllib.parse.quote_plus(url)
app.logger.info("Redirecting to login. Setting flow_state cookies")
@@ -417,15 +436,14 @@ def auth():
try:
login_request = HYDRA.login_request(challenge)
except hydra_client.exceptions.NotFound:
- app.logger.error("Not Found. Login request not found. challenge={0}".format(challenge))
- # TODO: Different error page?
- abort(404)
+ app.logger.error(f"Not Found. Login request not found. challenge={challenge}")
+ abort(404, description="Login request not found. Please try again.")
except hydra_client.exceptions.HTTPError:
- app.logger.error("Conflict. Login request has been used already. challenge={0}".format(challenge))
- # TODO: Different error page?
- abort(503)
+ app.logger.error(f"Conflict. Login request has been used already. challenge={challenge}")
+ abort(503, description="Login request already used. Please try again.")
# Authorize the user
+ # False positive: pylint: disable=no-member
redirect_to = login_request.accept(
profile['email'],
remember=True,
@@ -447,40 +465,38 @@ def consent():
challenge = request.args.get("consent_challenge")
if not challenge:
- # TODO: Better error handling/display?
- abort(403)
+ abort(403, description="Consent request required. Do not call this page directly")
try:
consent_request = HYDRA.consent_request(challenge)
except hydra_client.exceptions.NotFound:
- app.logger.error("Not Found. Consent request not found. challenge={0}".format(challenge))
- # TODO: Better error handling/display?
- abort(404)
+ app.logger.error(f"Not Found. Consent request {challenge} not found")
+ abort(404, description="Consent request does not exist. Please try again")
except hydra_client.exceptions.HTTPError:
- app.logger.error("Conflict. Consent request has been used already. challenge={0}".format(challenge))
- # TODO: Better error handling/display?
- abort(503)
+ app.logger.error(f"Conflict. Consent request {challenge} already used")
+ abort(503, description="Consent request already used. Please try again")
# Get information about this consent request:
+ # False positive: pylint: disable=no-member
app_name = consent_request.client.client_name
+ # False positive: pylint: disable=no-member
username = consent_request.subject
- # Get the related database object to get roles/access rights
- user = User.query.filter_by(email=username).first()
+ # Get the related user object
+ user = KratosUser.find_by_email(KRATOS_ADMIN, username)
if not user:
- app.logger.error("User not found in database: {0}".format(username))
- # TODO: Better error handling?
- abort(401)
+ app.logger.error(f"User not found in database: {username}")
+ abort(401, description="User not found. Please try again.")
# Get claims for this user, provided the current app
- claims = user.getClaims(app_name)
-
- # TODO: Check access / claims?
+ claims = user.get_claims(app_name)
- app.logger.info("Providing consent to %s for %s" % (app_name, username))
-
-
- app.logger.info("{0} was granted access to {1}".format(username, app_name))
+ # pylint: disable=fixme
+ # TODO: Need to implement checking claims here, once the backend for that is
+ # developed
+ app.logger.info(f"Providing consent to {app_name} for {username}")
+ app.logger.info(f"{username} was granted access to {app_name}")
+ # False positive: pylint: disable=no-member
return redirect(consent_request.accept(
grant_scope=consent_request.requested_scope,
grant_access_token_audience=consent_request.requested_access_token_audience,
@@ -495,16 +511,15 @@ def status():
Show if there is an user is logged in. If not shows: not-auth
"""
- auth = getAuth()
+ auth_status = get_auth()
- if auth:
- return auth['email']
- else:
- return "not-auth"
+ if auth_status:
+ return auth_status['email']
+ return "not-auth"
-def getAuth():
+def get_auth():
"""Checks if user is logged in
Queries the cookies. If an authentication cookie is found, it
checks with Kratos if the cookie is still valid. If so,
@@ -513,7 +528,7 @@ def getAuth():
"""
try:
- cookie = request.cookies.get('ory_kratos_session');
+ cookie = request.cookies.get('ory_kratos_session')
cookie = "ory_kratos_session=" + cookie
except TypeError:
app.logger.info("User not logged in or cookie corrupted")
@@ -529,13 +544,10 @@ def getAuth():
return profile
except ory_kratos_client.ApiException as error:
- app.logger.error("Exception when calling" +
- "V0alpha2Api->to_session(): %s\n" % error)
+ app.logger.error(f"Exception when calling V0alpha2Api->to_session(): {error}\n")
return False
if __name__ == '__main__':
app.run()
-
-
diff --git a/login/classes.py b/login/classes.py
new file mode 100644
index 0000000000000000000000000000000000000000..599545dd3dd38712316ed8b71b2544030328d39f
--- /dev/null
+++ b/login/classes.py
@@ -0,0 +1,17 @@
+
+"""Generic classes used by different parts of the application"""
+
+import urllib.request
+
+# Instead of processing the redirect, we return, so the application
+# can handle the redirect itself. This is needed to extract cookies
+# etc.
+class RedirectFilter(urllib.request.HTTPRedirectHandler):
+ """Overrides the standard redirect handler so it does not automatically
+ redirect. This allows for inspecting the return values before redirecting or
+ override the redirect action"""
+
+ # pylint: disable=too-many-arguments
+ # This amount of arguments is expected by the HTTPRedirectHandler
+ def redirect_request(self, req, fp, code, msg, headers, newurl):
+ return None
diff --git a/login/config.py b/login/config.py
index 57c900d61e21e897518fc331ef76451123615819..e13aecbb2baf50ea9e926a6e0031386e0487c956 100644
--- a/login/config.py
+++ b/login/config.py
@@ -1,10 +1,17 @@
+"""Config class imported by Flask to provide different enviroments with
+different debug level. It also sets the config object with the variables learned
+from the enviroment"""
+
import os
basedir = os.path.abspath(os.path.dirname(__file__))
-class Config(object):
+# Just provides config, not methods
+# pylint: disable=too-few-public-methods
+class Config():
+ """Flask Config object following the Flask specification"""
DEBUG = False
TESTING = False
CSRF_ENABLED = True
@@ -14,28 +21,33 @@ class Config(object):
SQLALCHEMY_TRACK_MODIFICATIONS = False
- PUBLIC_URL = os.environ['PUBLIC_URL'];
+ PUBLIC_URL = os.environ['PUBLIC_URL']
- HYDRA_ADMIN_URL = os.environ['HYDRA_ADMIN_URL'];
+ HYDRA_ADMIN_URL = os.environ['HYDRA_ADMIN_URL']
- KRATOS_ADMIN_URL = os.environ['KRATOS_ADMIN_URL'];
- KRATOS_PUBLIC_URL = os.environ['KRATOS_PUBLIC_URL'] + "/";
+ KRATOS_ADMIN_URL = os.environ['KRATOS_ADMIN_URL']
+ KRATOS_PUBLIC_URL = os.environ['KRATOS_PUBLIC_URL'] + "/"
class ProductionConfig(Config):
+ """Production config, which is the general config with DEBUGGING off"""
DEBUG = False
class StagingConfig(Config):
+ """Staging config which has debugging features on"""
DEVELOPMENT = True
DEBUG = True
class DevelopmentConfig(Config):
+ """Development config which has debugging features on"""
DEVELOPMENT = True
DEBUG = True
class TestingConfig(Config):
+ """Testing config which has debugging features off, but provides testing
+ flag"""
TESTING = True
diff --git a/login/exceptions.py b/login/exceptions.py
new file mode 100644
index 0000000000000000000000000000000000000000..bae5711c57fcf9dcc9a8fc96ccc03ef6cc77f6c7
--- /dev/null
+++ b/login/exceptions.py
@@ -0,0 +1,8 @@
+
+"""Custom exception handler to raise consistent exceptions, as different backend
+raise different exceptions"""
+
+class BackendError(Exception):
+ """The backend error is raised when interacting with
+ the backend fails or gives an unexpected result. The
+ error contains a oneliner description of the problem"""
diff --git a/login/kratos.py b/login/kratos.py
new file mode 100644
index 0000000000000000000000000000000000000000..eb3ea239373f1498f665e9d5d38240a48b10a16f
--- /dev/null
+++ b/login/kratos.py
@@ -0,0 +1,343 @@
+"""
+Implement the Kratos model to interact with kratos users
+"""
+
+import urllib.parse
+import urllib.request
+import json
+import re
+
+from typing import Dict
+from urllib.request import Request
+
+# Some imports commented out to satisfy pylint. They will be used once more
+# functions are migrated to this model
+from ory_kratos_client.model.admin_create_identity_body \
+ import AdminCreateIdentityBody
+from ory_kratos_client.model.admin_update_identity_body \
+ import AdminUpdateIdentityBody
+from ory_kratos_client.model.admin_create_self_service_recovery_link_body \
+ import AdminCreateSelfServiceRecoveryLinkBody
+from ory_kratos_client.rest import ApiException as KratosApiException
+
+from exceptions import BackendError
+from classes import RedirectFilter
+
+class KratosUser():
+ """
+ The User object, interact with the User. It both calls to Kratos as to
+ the database for storing and retrieving data.
+ """
+
+ api = None
+ __uuid = None
+ email = None
+ name = None
+ state = None
+ created_at = None
+ updated_at = None
+
+ def __init__(self, api, uuid = None):
+ self.api = api
+ self.state = 'active'
+ if uuid:
+ try:
+ obj = api.admin_get_identity(uuid)
+ if obj:
+ self.__uuid = uuid
+ try:
+ self.name = obj.traits['name']
+ except KeyError:
+ self.name = ""
+ self.email = obj.traits['email']
+ self.state = obj.state
+ self.created_at = obj.created_at
+ self.updated_at = obj.updated_at
+ except KratosApiException as error:
+ raise BackendError(f"Unable to get entry, kratos replied with: {error}") from error
+
+
+ def __repr__(self):
+ return f"\"{self.name}\" <{self.email}>"
+
+ @property
+ def uuid(self):
+ """Gets the protected UUID propery"""
+ return self.__uuid
+
+ def save(self):
+ """Saves this object into the kratos backend database. If the object
+ is new, it will create, otherwise update an entry.
+ :raise: BackendError is an error with Kratos happened.
+ """
+
+ # Traits are the "profile" values we will set, kratos will complain on
+ # empty values, so we check if "name" is set and only add it if so.
+ traits = {'email':self.email}
+
+ if self.name:
+ traits['name'] = self.name
+
+ # If we have a UUID, we are updating
+ if self.__uuid:
+ body = AdminUpdateIdentityBody(
+ schema_id="default",
+ state=self.state,
+ traits=traits,
+ )
+ try:
+ api_response = self.api.admin_update_identity(self.__uuid,
+ admin_update_identity_body=body)
+ except KratosApiException as error:
+ raise BackendError(f"Unable to save entry, kratos replied with:{error}") from error
+ else:
+
+ body = AdminCreateIdentityBody(
+ schema_id="default",
+ traits=traits,
+ )
+ try:
+ # Create an Identity
+ api_response = self.api.admin_create_identity(
+ admin_create_identity_body=body)
+ if api_response.id:
+ self.__uuid = api_response.id
+ except KratosApiException as error:
+ raise BackendError(f"Unable to save entry, kratos replied with:{error}") from error
+
+ def delete(self):
+ """Deletes the object from kratos
+ :raise: BackendError if Krator API call fails
+ """
+ if self.__uuid:
+ try:
+ self.api.admin_delete_identity(self.__uuid)
+ return True
+ except KratosApiException as error:
+ raise BackendError(
+ f"Unable to delete entry, kratos replied with: {error}"
+ ) from error
+
+ return False
+
+ @staticmethod
+ def find_by_email(api, email):
+ """Queries Kratos to find kratos ID for this given identifier
+ :param api: Kratos ADMIN API Object
+ :param email: Identifier to look for
+ :return: Return none or string with ID
+ """
+
+ kratos_id = None
+
+ # Get out user ID by iterating over all available IDs
+ data = api.admin_list_identities()
+ for kratos_obj in data.value:
+ # Unique identifier we use
+ if kratos_obj.traits['email'] == email:
+ kratos_id = str(kratos_obj.id)
+ return KratosUser(api, kratos_id)
+
+ return None
+
+ @staticmethod
+ def find_all(api):
+ """Queries Kratos to find all kratos users and return them
+ as a list of KratosUser objects
+ :return: Return list
+ """
+
+ kratos_id = None
+ return_list = []
+ # Get out user ID by iterating over all available IDs
+ data = api.admin_list_identities()
+ for kratos_obj in data.value:
+ kratos_id = str(kratos_obj.id)
+ return_list.append(KratosUser(api, kratos_id))
+
+ return return_list
+
+
+ @staticmethod
+ def extract_cookies(cookies):
+ """Extract session and CSRF cookie from a list of cookies.
+
+ Iterate over a list of cookies and extract the session
+ cookies required for Kratos User Panel UI
+
+ :param cookies: str[], list of cookies
+ :return: Cookies concatenated as string
+ :rtype: str
+ """
+
+ # Find kratos session cookie & csrf
+ cookie_csrf = None
+ cookie_session = None
+ for cookie in cookies:
+ search = re.match(r'ory_kratos_session=([^;]*);.*$', cookie)
+ if search:
+ cookie_session = "ory_kratos_session=" + search.group(1)
+ search = re.match(r'(csrf_token[^;]*);.*$', cookie)
+ if search:
+ cookie_csrf = search.group(1)
+
+ if not cookie_csrf or not cookie_session:
+ raise BackendError("Flow started, but expected cookies not found")
+
+ # Combined the relevant cookies
+ cookie = cookie_csrf + "; " + cookie_session
+ return cookie
+
+
+ def get_recovery_link(self):
+ """Call the kratos API to create a recovery URL for a kratos ID
+ :param: api Kratos ADMIN API Object
+ :param: kratos_id UUID of kratos object
+ :return: Return none or string with recovery URL
+ """
+
+ try:
+ # Create body request to get recovery link with admin API
+ body = AdminCreateSelfServiceRecoveryLinkBody(
+ expires_in="15m",
+ identity_id=self.__uuid
+ )
+
+ # Get recovery link from admin API
+ call = self.api.admin_create_self_service_recovery_link(
+ admin_create_self_service_recovery_link_body=body)
+
+ url = call.recovery_link
+ except KratosApiException:
+ return None
+ return url
+
+
+
+ def ui_set_password(self, api_url, recovery_url, password):
+ """Follow a Kratos UI sequence to set password
+ Kratos does not provide an interface to set a password directly. However
+ we still can set a password by following the UI sequence. To so so we
+ to follow the steps which are normally done in a browser once someone
+ clicks the recovery link.
+ :param: api_url URL to public endpoint of API
+ :param: recovery_url Recovery URL as generated by Kratos
+ :param: password Password
+ :raise: Exception with error message as first argument
+ :return: boolean True on success, False on failure (usualy password
+ to simple)
+ """
+
+ # Step 1: Open the recovery link and extract the cookies, as we need them
+ # for the next steps
+ try:
+ # We override the default Redirect handler with our custom handler to
+ # be able to catch the cookies.
+ opener = urllib.request.build_opener(RedirectFilter)
+ opener.open(recovery_url)
+ # If we do not have a 2xx status, urllib throws an error, as we "stopped"
+ # at our redirect, we expect a 3xx status
+ except urllib.error.HTTPError as req:
+ if req.status == 302:
+ cookies = req.headers.get_all('Set-Cookie')
+ url = req.headers.get('Location')
+ else:
+ raise BackendError('Unable to fetch recovery link') from req
+ else:
+ raise BackendError('Unable to fetch recovery link')
+
+ # Step 2: Extract cookies and data for next step. We expect to have an
+ # authorized session now. We need the cookies for followup calls
+ # to make changes to the account (set password)
+
+ # Get flow id
+ search = re.match(r'.*\?flow=(.*)', url)
+ if search:
+ flow = search.group(1)
+ else:
+ raise BackendError('No Flow ID found for recovery sequence')
+
+ # Extract cookies with helper function
+ cookie = self.extract_cookies(cookies)
+
+ # Step 3: Get the "UI", kratos expect us to call the API to get the UI
+ # elements which contains the CSRF token, which is needed when
+ # posting the password data
+ try:
+ url = api_url + "/self-service/settings/flows?id=" + flow
+
+ req = Request(url, headers={'Cookie':cookie})
+ opener = urllib.request.build_opener()
+
+ # Execute the request, read the data, decode the JSON, get the
+ # right CSRF token out of the decoded JSON
+ obj = json.loads(opener.open(req).read())
+ token = obj['ui']['nodes'][0]['attributes']['value']
+
+ except Exception as error:
+ raise BackendError("Unable to get password reset UI") from error
+
+
+ # Step 4: Post out password
+ url = api_url + "self-service/settings?flow=" + flow
+
+ # Create POST data as form data
+ data = {
+ 'method': 'password',
+ 'password': password,
+ 'csrf_token': token
+ }
+ data = urllib.parse.urlencode(data)
+ data = data.encode('ascii')
+
+ # POST the new password
+ try:
+ req = Request(url, data = data, headers={'Cookie':cookie}, method="POST")
+ opener = urllib.request.build_opener(RedirectFilter)
+ opener.open(req)
+ # If we do not have a 2xx status, urllib throws an error, as we "stopped"
+ # at our redirect, we expect a 3xx status
+ except urllib.error.HTTPError as req:
+ if req.status == 302:
+ return True
+ if req.status == 303:
+ # Something went wrong, usuall because the password is too
+ # simple. Kratos does not give a proper hint about the
+ # underlying error
+ return False
+ raise BackendError("Unable to set password by submitting form")
+
+ # Pylint complains about app not used. That is correct, but we will use that
+ # in the future. Ignore this error
+ # pylint: disable=unused-argument
+ def get_claims(self, app, mapping = None) -> Dict[str, Dict[str, str]]:
+ """Create openID Connect token
+ Use the userdata stored in the user object to create an OpenID Connect token.
+ The token returned by this function can be passed to Hydra,
+ which will store it and serve it to OpenID Connect Clients to retrieve user information.
+ If you need to relabel a field pass an array of tuples to rename_fields.
+ Example: getClaims('nextcloud', [("name", "username"),("roles", "groups")])
+
+ Attributes:
+ appname - Name or ID of app to connect to
+ mapping - Mapping of the fields
+
+ Returns:
+ OpenID Connect token of type dict
+ """
+
+ token = {
+ "name": self.email,
+ "preferred_username": self.email,
+ "email": self.email,
+ "roles": '',
+ }
+
+
+ # Relabel field names
+ if mapping:
+ for old_field_name, new_field_name in mapping:
+ token[new_field_name] = token[old_field_name]
+ del token[old_field_name]
+
+ return dict(id_token=token)
diff --git a/login/models.py b/login/models.py
index 5177edba9836d8637f31f3d36e0e58aad68bc448..07bb55822e68454d932e7c1f7967a9a9975fc4c6 100644
--- a/login/models.py
+++ b/login/models.py
@@ -1,62 +1,42 @@
+"""
+Implement different models used by Stackspin panel
+"""
-from app import db
-from sqlalchemy.dialects.postgresql import JSON
-
-from typing import Dict, List
-from sqlalchemy.orm import relationship
-from sqlalchemy import Column, Integer, String, ForeignKey, Boolean
+# We need this import at some point to hook up roles and users
+#from sqlalchemy.orm import relationship
+from sqlalchemy import Integer, String, ForeignKey, Boolean
+# pylint: disable=cyclic-import
+# This is based on the documentation of Flask Alchemy
+from app import db
-# User model
+# Pylint complains about too-few-public-methods. Methods will be added once
+# this is implemented.
+# pylint: disable=too-few-public-methods
class User(db.Model):
+ """
+ Represents a user in Kratos as well as this application's database.
+ """
id = db.Column(Integer, primary_key=True)
email = db.Column(String, unique=True)
kratos_id = db.Column(String, unique=True)
admin = db.Column(Boolean, default=False)
-# app_roles = relationship('AppRole', back_populates="users")
-
-
def __repr__(self):
return f"{self.id} <{self.email}>"
- def getClaims(self, app, mapping = []) -> Dict[str, Dict[str, str]]:
- """Create openID Connect token
- Use the userdata stored in the user object to create an OpenID Connect token.
- The token returned by this function can be passed to Hydra,
- which will store it and serve it to OpenID Connect Clients to retrieve user information.
- If you need to relabel a field pass an array of tuples to rename_fields.
- Example: getClaims('nextcloud', [("name", "username"),("roles", "groups")])
-
- Attributes:
- appname - Name or ID of app to connect to
- mapping - Mapping of the fields
-
- Returns:
- OpenID Connect token of type dict
- """
-
- token = {
- "name": self.email,
- "preferred_username": self.email,
- "email": self.email,
- "roles": '',
- }
-
- # Relabel field names
- for old_field_name, new_field_name in mapping:
- token[new_field_name] = token[old_field_name]
- del token[old_field_name]
- return dict(id_token=token)
-
-
-
-
-# App model
+# Pylint complains about too-few-public-methods. Methods will be added once
+# this is implemented.
+# pylint: disable=too-few-public-methods
class App(db.Model):
+ """
+ The App object, interact with the App database object. Data is stored in
+ the local database.
+ """
+
id = db.Column(Integer, primary_key=True)
name = db.Column(String())
@@ -65,9 +45,13 @@ class App(db.Model):
def __repr__(self):
return f"{self.id} <{self.name}>"
-
-# App role
+# Pylint complains about too-few-public-methods. Methods will be added once
+# this is implemented.
+# pylint: disable=too-few-public-methods
class AppRole(db.Model):
+ """
+ The AppRole object, stores the roles Users have on Apps
+ """
user_id = db.Column(Integer, ForeignKey('user.id'), primary_key=True)
app_id = db.Column(Integer, ForeignKey('app.id'),
@@ -77,5 +61,3 @@ class AppRole(db.Model):
# app = relationship("App")
role = db.Column(String)
-
-
diff --git a/pylintrc b/pylintrc
new file mode 100644
index 0000000000000000000000000000000000000000..0fd9caadf7a5736d5e5ab20949877a3d26e401f6
--- /dev/null
+++ b/pylintrc
@@ -0,0 +1,3 @@
+
+[TYPECHECK]
+generated-members=app.logger
diff --git a/test/lint/pylint/requirements.txt b/test/lint/pylint/requirements.txt
index 6684ea081ad050309f2375d1518e58142b5db46f..080f093f2c01740b40596563d4573399e1dc6d08 100644
--- a/test/lint/pylint/requirements.txt
+++ b/test/lint/pylint/requirements.txt
@@ -3,3 +3,4 @@ pylint
pylint-flask-sqlalchemy
pylint-flask
oic
+ory-kratos-client
diff --git a/test/sso_testapp/app.py b/test/sso_testapp/app.py
index b1f5c031ff58c055594eca82d60eec82b38cf0b5..bcb8e6418e67e5d74a77f49115f5e86d07ea58e7 100644
--- a/test/sso_testapp/app.py
+++ b/test/sso_testapp/app.py
@@ -41,6 +41,7 @@ def index():
if "state" not in session.keys():
state = rndstr()
session["state"] = state
+ # pylint: disable=consider-iterating-dictionary
if session["state"] not in sso.grant.keys() or not sso.grant[session["state"]].is_valid():
sso_auth_args["nonce"] = rndstr()
sso_auth_args["state"] = session["state"]