diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 8809b27431d85da64c5001ea4bd6173065fc7bbf..2cc9ba94e3859389ea73168cbe0a903da212c7a2 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -220,12 +220,12 @@ pylint:
- test/lint/pylint/requirements.txt
- .gitlab-ci.yml
-#pylint-lint:
-# stage: lint
-# variables:
-# GIT_STRATEGY: clone
-# GIT_DEPTH: 0
-# 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 .
+pylint-lint:
+ stage: lint
+ variables:
+ GIT_STRATEGY: clone
+ GIT_DEPTH: 0
+ 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 .
diff --git a/login/app.py b/login/app.py
index fd3076439da8ded2f7d2b107cd8671935fca927d..ebb56e6b13d2353195f67ff22bb4b9d7e944dd03 100644
--- a/login/app.py
+++ b/login/app.py
@@ -25,19 +25,15 @@ 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.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 exceptions import BackendError
+from kratos import KratosUser
# Initaliaze the FLASK app
@@ -154,6 +150,41 @@ app.cli.add_command(app_cli)
## CLI USER COMMANDS
+@user_cli.command('show')
+@click.argument('email')
+def show_user(email):
+ 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('updatename')
+@click.argument('email')
+@click.argument('name')
+def update_name_user(email, name):
+ user = KratosUser.find_by_email(KRATOS_ADMIN, email)
+ user.name = name
+ user.save()
+
+@user_cli.command('updateemail')
+@click.argument('email')
+@click.argument('newemail')
+def update_email_user(email, newemail):
+ user = KratosUser.find_by_email(KRATOS_ADMIN, email)
+ user.email = newemail
+ user.save()
+
+
+@user_cli.command('delete')
+@click.argument('email')
+def show_user(email):
+ user = KratosUser.find_by_email(KRATOS_ADMIN, email)
+ user.delete()
+
+
@user_cli.command('create')
@click.argument('email')
@@ -163,16 +194,22 @@ def create_user(email):
obj = User()
obj.email = email
- # Trying to create idenity
- try:
- body = AdminCreateIdentityBody(
- schema_id="default",
- traits={'email':email},
- ) # AdminCreateIdentityBody | (optional)
- kratos_obj = KRATOS_ADMIN.admin_create_identity(admin_create_identity_body=body)
- except ory_kratos_client.exceptions.ApiException as err:
- if err.status == 409:
- print("Conflict during creation of user. User already exists?")
+ # Load kratos object
+ user = KratosUser(KRATOS_ADMIN)
+ user.email = email
+ user.save()
+ user = KratosUser(email)
+
+# # Trying to create idenity
+# try:
+# body = AdminCreateIdentityBody(
+# schema_id="default",
+# traits={'email':email},
+# ) # AdminCreateIdentityBody | (optional)
+# kratos_obj = KRATOS_ADMIN.admin_create_identity(admin_create_identity_body=body)
+# except ory_kratos_client.exceptions.ApiException as err:
+# if err.status == 409:
+# print("Conflict during creation of user. User already exists?")
@@ -185,18 +222,8 @@ def create_user(email):
#
# - 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)
-
-
- obj.kratos_id = kratos_obj.id
+ obj.kratos_id = user.uuid
db.session.add(obj)
@@ -208,9 +235,10 @@ def create_user(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)
+ :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
"""
@@ -222,17 +250,16 @@ def setpassword_user(email, password):
# 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)
+ kratos_user = KratosUser.find_by_email(KRATOS_ADMIN, email)
# Get a recovery URL
- url = obj.get_recovery_link(KRATOS_ADMIN, kratos_id)
+ url = kratos_user.get_recovery_link()
# Execute UI sequence to set password, given we have a recovery URL
- result = obj.ui_set_password(app.config["KRATOS_PUBLIC_URL"], url, password)
+ result = kratos_user.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
@@ -249,15 +276,15 @@ def setpassword_user(email, password):
@user_cli.command('list')
def list_user():
app.logger.info("Listing users")
- users = User.query.all()
+ users = KratosUser.find_all(KRATOS_ADMIN)
for obj in users:
- print("Email: %s (admin: %s)" %(obj.email, obj.admin))
+ print(obj)
-@user_cli.command('delete',)
+@user_cli.command('delete2',)
@click.argument('email')
-def delete_user(email):
+def delete2_user(email):
app.logger.info("Trying to delete user: {0}".format(email))
obj = User.query.filter_by(email=email).first()
diff --git a/login/exceptions.py b/login/exceptions.py
index ef723b537bfaeec3b4a79a87f764db861de869d7..4b0747a33ccf538f0bf5a87b361b6c86fb0235cb 100644
--- a/login/exceptions.py
+++ b/login/exceptions.py
@@ -1,4 +1,6 @@
class BackendError(Exception):
- """Base class for other exceptions"""
+ """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"""
pass
diff --git a/login/kratos.py b/login/kratos.py
index 8e83eb839552e62e95b46510cd06e733700b4b91..57743858270b3ee940037e4590c879a5f7cb3dbe 100644
--- a/login/kratos.py
+++ b/login/kratos.py
@@ -17,8 +17,8 @@ from sqlalchemy import 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.admin_create_identity_body \
+ import AdminCreateIdentityBody
from ory_kratos_client.model.admin_update_identity_body \
import AdminUpdateIdentityBody
@@ -75,6 +75,10 @@ class KratosUser():
def save(self):
+
+ # Kratos API expect for state an IdentifyState object, while
+ # during updating, it expects a string.
+
if self.uuid:
body = AdminUpdateIdentityBody(
schema_id="default",
@@ -88,15 +92,21 @@ class KratosUser():
except ory_kratos_client.ApiException as e:
print("Exception when calling V0alpha2Api->admin_update_identity: %s\n" % e)
else:
+ traits = {'email':self.email}
+
+ if self.name:
+ traits['name'] = self.name
+
body = AdminCreateIdentityBody(
schema_id="default",
- state=self.state,
- traits={'email':self.email, 'name':self.name},
+ traits=traits
)
try:
- # Update an Identity
- api_response = self.api.admin_create_identity(self.uuid,
+ # 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 ory_kratos_client.ApiException as e:
print("Exception when calling V0alpha2Api->admin_update_identity: %s\n" % e)
diff --git a/login/models.py b/login/models.py
index 502b710663704a7b374c5bd96930318841a5e576..933c270dac2e039ceae90166b256283612103e68 100644
--- a/login/models.py
+++ b/login/models.py
@@ -82,173 +82,6 @@ class User(db.Model):
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):
- """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=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 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 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 BackendError("Flow started, but expected cookies not found")
-
- # Combined the relevant cookies
- cookie = cookie_csrf + "; " + cookie_session
- return cookie
-
- 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.