Merge pull request #23 from LifeguardSystem/basic-auth

implementation of basic authentication

Author: diegorubin

README.md

@@ -104,3 +104,18 @@ To see all settings avaiable run command:
 
 `lifeguard -d`
 
+
+## Authentication
+
+### Builtin Methods
+
+#### Basic Authentication
+
+Set users in lifeguard context like in the example:
+
+```
+# in lifeguard_settings.py
+def setup(lifeguard_context):
+    lifeguard_context.auth_method = BASIC_AUTH_METHOD
+    lifeguard_context.users = [{"username": "test", "password": "pass"}]
+```

lifeguard/auth.py

@@ -0,0 +1,35 @@
+"""
+Auth methods
+"""
+import base64
+
+from flask import Response as FlaskResponse
+from flask import request as flask_request
+
+from lifeguard.context import LIFEGUARD_CONTEXT
+
+BASIC_AUTH_METHOD = "basic_auth"
+
+
+def basic_auth_login(authorization_header):
+    """
+    Basic auth method
+    """
+    encoded_uname_pass = authorization_header.split()[-1]
+    username, password = base64.b64decode(encoded_uname_pass).decode("utf-8").split(":")
+    return LIFEGUARD_CONTEXT.valid_user(username, password)
+
+
+def basic_auth_login_required_implementation(args, kwargs, function):
+    """
+    Baic Auth Login Implementation
+    """
+    authorization_header = flask_request.headers.get("Authorization")
+    if authorization_header and basic_auth_login(authorization_header):
+        return function(*args, **kwargs)
+    response = FlaskResponse()
+    response.headers["WWW-Authenticate"] = "Basic"
+    return response, 401
+
+
+AUTHENTICATION_METHODS = {BASIC_AUTH_METHOD: basic_auth_login_required_implementation}

lifeguard/context.py

@@ -1,9 +1,16 @@
+"""
+Lifeguard Context Implementation
+"""
+
+
 class LifeguardContext:
     """
     Lifeguard Context
     """
 
     def __init__(self):
+        self._users = []
+        self._auth_method = None
         self._only_settings = False
         self._alert_email_template = """From: [[SMTP_USER]]
 To: [[RECEIVERS]]
@@ -51,5 +58,43 @@ def alert_email_template(self, value):
         """
         self._alert_email_template = value
 
+    @property
+    def auth_method(self):
+        """
+        Getter for auth method
+        """
+        return self._auth_method
+
+    @auth_method.setter
+    def auth_method(self, value):
+        """
+        Setter for auth method
+        """
+        self._auth_method = value
+
+    @property
+    def users(self):
+        """
+        Getter for users
+        """
+        return self._users
+
+    @users.setter
+    def users(self, value):
+        """
+        Setter for users
+        """
+        self._users = value
+
+    def valid_user(self, username, password):
+        """
+        Check if user is valid
+        """
+        return [
+            user
+            for user in self.users
+            if user["username"] == username and user["password"] == password
+        ]
+
 
 LIFEGUARD_CONTEXT = LifeguardContext()

lifeguard/controllers.py

@@ -8,6 +8,8 @@
 from flask import Response as FlaskResponse
 from flask import request as flask_request
 
+from lifeguard.auth import AUTHENTICATION_METHODS
+from lifeguard.context import LIFEGUARD_CONTEXT
 from lifeguard.logger import lifeguard_logger as logger
 from lifeguard.settings import LIFEGUARD_DIRECTORY
 
@@ -109,10 +111,27 @@ def data(self, value):
         self._data = value
 
 
+def login_required(function):
+    """
+    Decorator for login
+    """
+
+    @wraps(function)
+    def wrapped(*args, **kwargs):
+
+        if LIFEGUARD_CONTEXT.auth_method in AUTHENTICATION_METHODS:
+            return AUTHENTICATION_METHODS[LIFEGUARD_CONTEXT.auth_method](
+                args, kwargs, function
+            )
+        return function(*args, **kwargs)
+
+    return wrapped
+
+
 def register_custom_controller(path, function, options):
     endpoint = options.pop("endpoint", function.__name__)
     custom_controllers.add_url_rule(
-        path, endpoint, configure_controller(function), **options
+        path, endpoint, configure_controller(login_required(function)), **options
     )
 
 

lifeguard/server.py

@@ -4,7 +4,7 @@
 
 from flask import Flask, make_response
 
-from lifeguard.controllers import custom_controllers
+from lifeguard.controllers import custom_controllers, login_required
 from lifeguard.logger import lifeguard_logger as logger
 from lifeguard.repositories import ValidationRepository
 from lifeguard.validations import VALIDATIONS, ValidationResponseEncoder
@@ -19,13 +19,15 @@ def make_json_response(content):
 
 
 @APP.route("/lifeguard/validations/<validation>", methods=["GET"])
+@login_required
 def get_validation(validation):
     repository = ValidationRepository()
     result = repository.fetch_last_validation_result(validation)
     return make_json_response(ValidationResponseEncoder().encode(result))
 
 
 @APP.route("/lifeguard/validations/<validation>/execute", methods=["POST"])
+@login_required
 def execute_validation(validation):
     try:
         result = VALIDATIONS[validation]["ref"]()

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name="lifeguard",
-    version="0.0.21",
+    version="0.0.22",
     url="https://github.com/LifeguardSystem/lifeguard",
     author="Diego Rubin",
     author_email="[email protected]",

tests/test_auth.py

@@ -0,0 +1,96 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+from lifeguard.auth import (
+    basic_auth_login,
+    basic_auth_login_required_implementation,
+    AUTHENTICATION_METHODS,
+)
+from lifeguard.context import LifeguardContext
+
+lifeguard_context = LifeguardContext()
+lifeguard_context.users = [{"username": "test", "password": "pass"}]
+
+
+class TestAuth(unittest.TestCase):
+    @patch("lifeguard.auth.LIFEGUARD_CONTEXT", lifeguard_context)
+    def test_auth_valid_user(self):
+        header = "Basic dGVzdDpwYXNz"
+
+        self.assertListEqual(
+            basic_auth_login(header), [{"username": "test", "password": "pass"}]
+        )
+
+    @patch("lifeguard.auth.LIFEGUARD_CONTEXT", lifeguard_context)
+    def test_auth_header(self):
+        header = "Basic YTpi dGVzdDpwYXNz"
+
+        self.assertListEqual(
+            basic_auth_login(header), [{"username": "test", "password": "pass"}]
+        )
+
+    def test_auth_invalid_user(self):
+        header = "Basic YTpi"
+
+        self.assertFalse(basic_auth_login(header), [])
+
+    @patch("lifeguard.auth.FlaskResponse")
+    @patch("lifeguard.auth.flask_request", spec={})
+    @patch("lifeguard.auth.basic_auth_login")
+    def test_basic_auth_login_required_implementation_call_function(
+        self, mock_login, mock_request, _mock_response
+    ):
+        args = MagicMock(name="args")
+        kwargs = MagicMock(name="kwargs")
+        function = MagicMock(name="function")
+
+        mock_request.headers = MagicMock(name="headers")
+        mock_request.headers.get.return_value = "returned_get"
+
+        mock_login.return_value = True
+
+        function.return_value = False
+
+        self.assertFalse(
+            basic_auth_login_required_implementation(args, kwargs, function)
+        )
+
+        function.assert_called()
+        mock_request.headers.get.assert_called_with("Authorization")
+        mock_login.assert_called_with("returned_get")
+
+    @patch("lifeguard.auth.FlaskResponse")
+    @patch("lifeguard.auth.flask_request", spec={})
+    @patch("lifeguard.auth.basic_auth_login")
+    def test_basic_auth_login_required_implementation_return_401(
+        self, mock_login, mock_request, mock_response
+    ):
+        args = MagicMock(name="args")
+        kwargs = MagicMock(name="kwargs")
+        function = MagicMock(name="function")
+
+        mock_request.headers = MagicMock(name="headers")
+        mock_request.headers.get.return_value = "returned_get"
+
+        mock_login.return_value = False
+
+        response = MagicMock(name="response")
+        response.headers = {}
+        mock_response.return_value = response
+
+        self.assertEqual(
+            basic_auth_login_required_implementation(args, kwargs, function),
+            (response, 401),
+        )
+
+        self.assertDictEqual(response.headers, {"WWW-Authenticate": "Basic"})
+
+        function.assert_not_called()
+        mock_request.headers.get.assert_called_with("Authorization")
+        mock_login.assert_called_with("returned_get")
+
+    def test_authentication_methods(self):
+        self.assertEqual(
+            AUTHENTICATION_METHODS["basic_auth"],
+            basic_auth_login_required_implementation,
+        )

tests/test_controllers.py

@@ -8,11 +8,15 @@
     configure_controller,
     render_template,
     Request,
+    login_required,
 )
 from tests.fixtures.controllers.hello_controller import hello
 
 mock_flask_request = MagicMock(name="flask_request")
 
+MOCK_BASIC_AUTH = MagicMock(name="basic_auth")
+MOCK_AUTHENTICATION_METHODS = {"basic_auth": MOCK_BASIC_AUTH}
+
 
 class TestControllers(unittest.TestCase):
     @patch("lifeguard.controllers.LIFEGUARD_DIRECTORY", "tests/fixtures")
@@ -146,3 +150,29 @@ def test_request_proxy(self):
         self.assertEqual("method", request.method)
         self.assertEqual("args", request.args)
         self.assertEqual("values", request.values)
+
+    @patch("lifeguard.controllers.AUTHENTICATION_METHODS", MOCK_AUTHENTICATION_METHODS)
+    @patch("lifeguard.controllers.LIFEGUARD_CONTEXT")
+    def test_login_required_wrap_controller(self, mock_lifeguard_context):
+
+        mock_lifeguard_context.auth_method = "basic_auth"
+        MOCK_AUTHENTICATION_METHODS["basic_auth"].return_value = False
+
+        a_function = MagicMock(name="a_function")
+        wrapped = login_required(a_function)
+
+        self.assertFalse(wrapped())
+        MOCK_AUTHENTICATION_METHODS["basic_auth"].assert_called_with((), {}, a_function)
+
+    @patch("lifeguard.controllers.AUTHENTICATION_METHODS", MOCK_AUTHENTICATION_METHODS)
+    @patch("lifeguard.controllers.LIFEGUARD_CONTEXT")
+    def test_login_required_not_wrap_controller(self, mock_lifeguard_context):
+
+        mock_lifeguard_context.auth_method = None
+
+        a_function = MagicMock(name="a_function")
+        a_function.return_value = 1
+        wrapped = login_required(a_function)
+
+        self.assertEqual(wrapped(), 1)
+        MOCK_AUTHENTICATION_METHODS["basic_auth"].assert_not_called()