diff --git a/login/app.py b/login/app.py
index 87d07581c5fa6488ccf7e694a3d843a57f7976dc..9dad50b8c4bce7bafb96e3f5f36dc71c28a374be 100644
--- a/login/app.py
+++ b/login/app.py
@@ -4,12 +4,13 @@
import pprint
import logging
import os
-import click
import urllib.parse
import urllib.request
import json
import re
+import click
from urllib.request import Request
+from classes import RedirectFilter
# Flask
from flask import abort, Flask, redirect, request, render_template
@@ -39,6 +40,7 @@ from ory_kratos_client.model.self_service_recovery_flow import SelfServiceRecove
from ory_kratos_client.model.identity_state import IdentityState
+from exceptions import *
# Initaliaze the FLASK app
app = Flask(__name__,
@@ -156,10 +158,11 @@ app.cli.add_command(app_cli)
@user_cli.command('create')
@click.argument('email')
-@click.argument('password')
-def create_user(email, password):
+def create_user(email):
app.logger.info("Creating user with email: ({0})".format(email))
+ obj = User()
+ obj.email = email
# Trying to create idenity
try:
@@ -173,150 +176,6 @@ def create_user(email, password):
print("Conflict during creation of user. User already exists?")
- # 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 arround 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
-
-
- # Step 1: Get all recovery link for this user account
- url = ""
- try:
- # Get out user ID by iterating over all available IDs
- data = KRATOS_ADMIN.admin_list_identities()
- for id in data.value:
- # Unique identifier we use
- if (id.traits['email'] == email):
- kratos_id = id.id
-
- # Create body request to get recovery link with admin API
- body = AdminCreateSelfServiceRecoveryLinkBody(
- expires_in="15m",
- identity_id=str(kratos_id)
- )
-
- # Get recovery link from admin API
- api = KRATOS_ADMIN.admin_create_self_service_recovery_link(
- admin_create_self_service_recovery_link_body=body)
-
- url = api.recovery_link
- except:
- print("Unable to find user and/or create recovery link")
- return False
-
-
- # Step 2: 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)
- req = opener.open(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 http:
- cookies = http.headers.get_all('Set-Cookie')
- url = http.headers.get('Location')
- else:
- print("Unable to follow recovery link and get session cookies")
- return False
-
-
- # Step 3: 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
- m = re.match(r'.*\?flow=(.*)', url)
- if m:
- flow = m.group(1)
- else:
- print("Unable to extract flow ID")
- return False
-
- # Find kratos session cookie & csrf
- for cookie in cookies:
- m = re.match(r'ory_kratos_session=([^;]*);.*$', cookie)
- if m:
- session = "ory_kratos_session=" + m.group(1)
- m = re.match(r'(csrf_token[^;]*);.*$', cookie)
- if m:
- csrf = m.group(1)
-
- # Combined the relevant cookies
- cookie = csrf + "; " + session;
-
-
- # Step 4: Get the "UI", kratos expect us to call the API to get the UI
- # elements which inclused the CSRF token, which is needed when
- # posting the password data
- try:
- url = app.config["KRATOS_PUBLIC_URL"]
- url = 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
- http = opener.open(req)
- data = http.read()
- obj = json.loads(data)
- token = obj['ui']['nodes'][0]['attributes']['value']
-
- except:
- print("Unable to request password reset and get CSRF token")
- return False
-
-
- # Step 5: Post out password
-
- url = app.config["KRATOS_PUBLIC_URL"]
- url = 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)
- http = 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 http:
- if http.status == 302:
- print("Password is set!")
- elif http.status == 303:
- print("Password not set. To simple?")
- else:
- print("Password not set, unknown error")
-
- else:
- print("Unable to follow recovery link and get session cookies")
- return False
-
-
-
-
-
-
-
-@user_cli.command('invite')
-@click.argument('email')
-def invite_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
@@ -345,6 +204,49 @@ def invite_user(email):
db.session.commit()
+@user_cli.command('setpassword')
+@click.argument('email')
+@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: boolean true on success, if not set (too weak)
+ :raise: exception if unexepted error happens
+ """
+
+ app.logger.info("Creating user with email: ({0})".format(email))
+
+ # 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 arround 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
+
+ obj = User()
+
+ try:
+ # Get the ID of the user
+ kratos_id = obj.find_kratos_id(KRATOS_ADMIN, email)
+
+ # Get a recovery URL
+ url = obj.get_recovery_link(KRATOS_ADMIN, kratos_id)
+
+ # Execute UI sequence to set password, given we have a recovery URL
+ result = obj.ui_set_password(app.config["KRATOS_PUBLIC_URL"], url, password)
+ except BackendError as error:
+ app.logger.error(f"Error while setting password: {error}")
+ return False
+
+ if result:
+ app.logger.info("Success setting password")
+ else:
+ app.logger.error("Failed to set password. Password too weak?")
+
+ return result
+
+
+
@user_cli.command('list')
def list_user():
app.logger.info("Listing users")
@@ -493,7 +395,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
@@ -511,7 +413,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")
@@ -525,7 +427,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
@@ -536,8 +438,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")
@@ -633,7 +535,7 @@ def consent():
abort(401)
# Get claims for this user, provided the current app
- claims = user.getClaims(app_name)
+ claims = user.get_claims(app_name)
# TODO: Check access / claims?
@@ -700,10 +602,3 @@ if __name__ == '__main__':
app.run()
-# 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):
- def redirect_request(self, req, fp, code, msg, hdrs, newurl):
- return None
-
diff --git a/login/models.py b/login/models.py
index 5177edba9836d8637f31f3d36e0e58aad68bc448..bfd6b09b86d3ac5ee3ea9466f7b1a91085abbadf 100644
--- a/login/models.py
+++ b/login/models.py
@@ -1,33 +1,62 @@
+"""
+Implement different models used by Stackspin panel
+"""
-from app import db
-from sqlalchemy.dialects.postgresql import JSON
+import urllib.parse
+import urllib.request
+import json
+import re
+
+from typing import Dict
+from urllib.request import Request
-from typing import Dict, List
+# 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
-from sqlalchemy.orm import relationship
-from sqlalchemy import Column, Integer, String, ForeignKey, Boolean
+# 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.submit_self_service_recovery_flow_body \
+# import SubmitSelfServiceRecoveryFlowBody
+#from ory_kratos_client.model.self_service_recovery_flow \
+# import SelfServiceRecoveryFlow
+#from ory_kratos_client.model.identity_state \
+# import IdentityState
+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
+from app import db
# User model
class User(db.Model):
+ """
+ The User object, interact with the User. It both calls to Kratos as to
+ the database for storing and retrieving data.
+ """
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]]:
+ # 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.
+ If you need to relabel a field pass an array of tuples to rename_fields.
Example: getClaims('nextcloud', [("name", "username"),("roles", "groups")])
Attributes:
@@ -45,18 +74,190 @@ class User(db.Model):
"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]
+ 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)
+ @staticmethod
+ def find_kratos_id(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 kratos_id
+
+
+ @staticmethod
+ def get_recovery_link(api, kratos_id):
+ """Queries Kratos to find kratos ID for this given identifier
+ :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=str(kratos_id)
+ )
+
+ # Get recovery link from admin API
+ call = 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
+
+
+ @staticmethod
+ def extract_cookies(cookies):
+ """Extract required session cookies our of cookie array
+ Given an cookie array, iterate over the array and extrat the session
+ cookies required for Kratos User Panel UI
+ :param: cookies List of cookies
+ :return: string Cookies as string
+ """
+
+ # 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 Exception("Flow started, but expected cookies not found")
+ # Combined the relevant cookies
+ cookie = cookie_csrf + "; " + cookie_session
+ return cookie
-# App model
+ 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: boolen 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 inclused 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 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 +266,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 +282,3 @@ class AppRole(db.Model):
# app = relationship("App")
role = db.Column(String)
-
-