.gitlab-ci.yml

 stages:
   - build
-  - build-testimages
+  - build-test-images
   - application-test
 
 consent-provider:
@@ -30,7 +30,7 @@ login-provider:
       - .gitlab-ci.yml
 
 login-provider-mock:
-  stage: build-testimages
+  stage: build-test-images
   image:
     name: gcr.io/kaniko-project/executor:debug
     entrypoint: [""]
@@ -43,7 +43,7 @@ login-provider-mock:
       - .gitlab-ci.yml
 
 login-provider-test:
-  stage: build-testimages
+  stage: build-test-images
   image:
     name: gcr.io/kaniko-project/executor:debug
     entrypoint: [""]
@@ -55,12 +55,20 @@ login-provider-test:
       - login_provider/**/*
       - .gitlab-ci.yml
 
+unittest-login:
+  stage: application-test
+  image: ${CI_REGISTRY_IMAGE}/login_provider_ci_test:${CI_COMMIT_REF_NAME}
+  script:
+    - cd login_provider
+    - python3 -m unittest discover
+
 behave-login:
   stage: application-test
   variables:
     TESTUSER_USERNAME: "admin"
     TESTUSER_EMAIL: "admin@example.net"
     TESTUSER_PASSWORD: "password"
+    FLASK_ENV: "development"
   services:
     - name: ${CI_REGISTRY_IMAGE}/login_provider_mock:${CI_COMMIT_REF_NAME}
       alias: login_provider

login_provider/Dockerfile

@@ -11,6 +11,8 @@ EXPOSE 5000
 
 ENV FLASK_ENV production
 ENV FLASK_RUN_HOST 0.0.0.0
-ENV HYDRA_ADMIN_URL http://localhost:4445
+ENV FLASK_RUN_PORT 5000
+ENV HYDRA_ADMIN_URL http://localhost:444
+ENV GRAPHQL_URL http://localhost:5002/graphql
 
 CMD [ "flask", "run" ]

login_provider/README.md

+# Configuration
+
+To enable the `debug` mode, set the environment variable `FLASK_ENV` to `development`.
+
+```
+export FLASK_ENV=development
+# or
+docker login-provider:latest build . && docker run -e FLASK_ENV=development login-provider
+```
+
+You can do the same with the following variables.
+
+ * **FLASK_SECRET_KEY** A secret key that will be used for securely signing the session cookie.
+ * **FLASK_RUN_HOST** IP Address that the server will open a socket on.
+   *Default*: 0.0.0.0
+ * **FLASK_RUN_PORT** Port of the socket that the server will listen on.
+   *Default*: 5000
+ * **GRAPHQL_URL** URL to the server that runs the graphql backend API
+   *Default*: http://localhost:5002/graphql
+ * **HYDRA_ADMIN_URL** URl to the Hydra admin server
+   *Default*: http://localhost:4445
+   

login_provider/app.py

@@ -4,13 +4,14 @@ from hydra_client import HydraAdmin
 from flask_login import login_user, logout_user, LoginManager, login_required, current_user
 from db import User
 from forms import LoginForm, LogoutForm
+from helper import is_safe_url
 
 HYDRA_ADMIN_URL = environ['HYDRA_ADMIN_URL']
-hydra = HydraAdmin(HYDRA_ADMIN_URL)
+HYDRA = HydraAdmin(HYDRA_ADMIN_URL)
 
 app = Flask(__name__)
-app.config['SECRET_KEY'] = urandom(16)
-app.debug = True if "DEBUG" in environ and environ["DEBUG"] else False
+if "FLASK_SECRET_KEY" not in environ:
+    app.config['SECRET_KEY'] = urandom(16)
 
 login_manager = LoginManager()
 login_manager.init_app(app)
@@ -20,19 +21,17 @@ login_manager.login_view = "login"
 def user_loader(username):
     user = User(username)
     if not user.active:
-        return
+        return None
     return user
 
 @app.route('/', methods=['GET'])
 @login_required
 def home():
-    logout_form = LogoutForm()
     challenge = request.args.get("login_challenge")
     if not challenge:
-        return render_template('home.html', email=current_user.email, logout_form=logout_form)
-    else:
-        redirect_to = hydra.login_request(challenge).accept(current_user.email)
-        return redirect(redirect_to)
+        return render_template('home.html', email=current_user.email, logout_form=LogoutForm())
+    redirect_to = HYDRA.login_request(challenge).accept(current_user.email)
+    return redirect(redirect_to)
 
 @app.route('/login', methods=['GET', 'POST'])
 def login():
@@ -43,19 +42,11 @@ def login():
             login_user(user)
         next_url = login_form.next_url.data
         if not is_safe_url(next_url):
-            return abort(400)
+            return redirect(url_for('home'))
         return redirect(next_url or url_for('home'))
     login_form.next_url.data = request.args.get('next')
     return render_template('login.html', login_form=login_form)
 
-def is_safe_url(url):
-    print(url)
-    safe = True if url == "" else False
-    safe = True if url == "/" or safe else False
-    safe = True if url[:18] == "/?login_challenge=" \
-                   and url[18:].isalnum() or safe else False
-    return safe
-
 @app.route('/logout', methods=['POST'])
 def logout():
     logout_form = LogoutForm()

login_provider/db.py

@@ -7,6 +7,7 @@ from json import loads
 GRAPHQL_URL = environ['GRAPHQL_URL']
 graphql_client = GraphQLClient(GRAPHQL_URL)
 
+
 class User(UserMixin):
     def __init__(self, username):
         self.id = username
@@ -15,23 +16,28 @@ class User(UserMixin):
         self._load_remote_user_info()
 
     def _load_remote_user_info(self):
-        querystring = '''{{
-        getUser(username: "{0}"){{
+        querystring = '''{
+        getUser(username: $username){
             email,
             active
-            }}}}'''.format(self.username).strip()
-        result = loads(graphql_client.execute(querystring))
+            }}'''
+        result = loads(graphql_client.execute(querystring, {'username': self.username}))
         if "data" in result:
             self.active = result["data"]["getUser"]["active"]
             self.email = result["data"]["getUser"]["email"]
 
     def _verify_password(self, password):
-        querystring = '''{{
+        querystring = '''{
         verifyPassword(
-            username: "{0}",
-            password: "{1}")
-            }}'''.format(self.username, password).strip()
-        result = loads(graphql_client.execute(querystring))
+            username: $username,
+            password: $password)
+            }'''
+        result = loads(
+            graphql_client.execute(querystring, {
+                'username': self.username,
+                'password': password
+                })
+            )
         verified = False
         if "data" in result:
             verified = result["data"]["verifyPassword"]

login_provider/helper.py

+import re
+
+
+def is_safe_url(url):
+    """Checks if a url is safe
+
+    Check if a url is safe to be used in redirects. This function is used whenever the user
+    passes a redirect url to the application. In case of the login process the user passes
+    a url to the application via a HTTP-GET variable, namely `next`. Once the user successfully
+    authenticated, the url is used by the server to redirect to the page the user initially
+    requested. The url validation prevents attacks where an attacker creates links that
+    redirect users to malicious urls once they are loged in.
+    example: http://login-provider/login?next=malicious\.org/
+
+    Args:
+        url: Url that needs to be validated
+
+    Returns:
+        True if the url is trusted. False if not.
+    """
+    safe_urls = [
+        r"^[/]*$",                             # Home page
+        r"^/\?login_challenge=[a-z|A-Z|0-9]+$" # Login challenge with alphanumeric code
+        ]
+    for safe_url in safe_urls:
+        if re.fullmatch(safe_url, url) is not None:
+            return True
+    return False

login_provider/test/behave/mocks/db.py

@@ -5,7 +5,7 @@ USERNAME = environ["TESTUSER_USERNAME"]
 PASSWORD = environ["TESTUSER_PASSWORD"]
 EMAIL = environ["TESTUSER_EMAIL"]
 
-users = {USERNAME: {"password": PASSWORD, "email": EMAIL, "active": True}}
+USERS = {USERNAME: {"password": PASSWORD, "email": EMAIL, "active": True}}
 
 class User(UserMixin):
     def __init__(self, username):
@@ -15,12 +15,12 @@ class User(UserMixin):
         self._load_remote_user_info()
 
     def _load_remote_user_info(self):
-        if self.username in users:
-            self.active = users[self.username]["active"]
-            self.email = users[self.username]["email"]
+        if self.username in USERS:
+            self.active = USERS[self.username]["active"]
+            self.email = USERS[self.username]["email"]
 
     def _verify_password(self, password):
-        return users[self.username]["password"] == password
+        return USERS[self.username]["password"] == password
 
     def authenticate(self, password):
         return self.active and self._verify_password(password)

login_provider/test/test_helper_functions.py

+import unittest
+from helper import is_safe_url
+
+class UnitTests(unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def test_safe_urls(self):
+        self.assertTrue(is_safe_url("/"))
+        self.assertTrue(is_safe_url("/?login_challenge=9a8s9da8s9dhahsda"))
+        self.assertFalse(is_safe_url("/malicious"))
+        self.assertFalse(is_safe_url("/?login_challenge=Not_alpha_numeric"))