Merge pull request #22 from nationalarchives/make-current-poc-routes-…

…protected

Make current poc routes protected

app/__init__.py

@@ -1,3 +1,5 @@
+import os
+
 import boto3
 from flask import Flask
 from flask_assets import Bundle, Environment
@@ -28,18 +30,19 @@ def null_to_dash(value):
     return value
 
 
+# use only for local development
+if os.environ.get("DEFAULT_AWS_PROFILE"):
+    boto3.setup_default_session(
+        profile_name=os.environ.get("DEFAULT_AWS_PROFILE")
+    )
+
+
 def create_app(config_class=Config):
     app = Flask(__name__, static_url_path="/assets")
-    app.config.from_object(config_class)
+    app.config.from_object(config_class())
 
     force_https = False if app.config["TESTING"] else True
 
-    # use only for local development
-    if app.config["DEFAULT_AWS_PROFILE"]:
-        boto3.setup_default_session(
-            profile_name=app.config["DEFAULT_AWS_PROFILE"]
-        )
-
     app.jinja_env.lstrip_blocks = True
     app.jinja_env.trim_blocks = True
     app.jinja_env.filters["null_to_dash"] = null_to_dash

app/main/authorize/keycloak_login_required_decorator.py

@@ -0,0 +1,82 @@
+from functools import wraps
+
+import keycloak
+from flask import current_app, flash, redirect, session, url_for
+
+
+def access_token_login_required(view_func):
+    """
+    Decorator that checks if the user is logged in via Keycloak and has access to AYR.
+
+    This decorator is typically applied to view functions that require authentication via Keycloak
+    and access to the AYR application. It checks for the presence of an access token in the session,
+    verifies the token's validity, and checks if the user belongs to the AYR user group in Keycloak.
+
+    Args:
+        view_func (function): The view function to be wrapped.
+
+    Returns:
+        function: The wrapped view function.
+
+    If the user is not authenticated or does not have access, this decorator redirects to the login page
+    or the main index and displays a flash message accordingly.
+
+    Configuration options for Keycloak, such as the client ID, realm name, base URI, and client secret,
+    are expected to be set in the Flask application configuration.
+
+    When the application is running in testing mode and the 'FORCE_AUTHENTICATION_FOR_IN_TESTING' config
+    option is not set, the decorator allows unauthenticated access to facilitate testing.
+
+    Example:
+        @app.route('/protected')
+        @access_token_login_required
+        def protected_route():
+            return 'Access granted'
+    """
+
+    @wraps(view_func)
+    def decorated_view(*args, **kwargs):
+        if current_app.config["TESTING"] and not current_app.config.get(
+            "FORCE_AUTHENTICATION_FOR_IN_TESTING"
+        ):
+            return view_func(*args, **kwargs)
+
+        access_token = session.get("access_token")
+        if not access_token:
+            return redirect(url_for("main.login"))
+
+        keycloak_openid = keycloak.KeycloakOpenID(
+            server_url=current_app.config["KEYCLOAK_BASE_URI"],
+            client_id=current_app.config["KEYCLOAK_CLIENT_ID"],
+            realm_name=current_app.config["KEYCLOAK_REALM_NAME"],
+            client_secret_key=current_app.config["KEYCLOAK_CLIENT_SECRET"],
+        )
+
+        decoded_token = keycloak_openid.introspect(access_token)
+
+        if not decoded_token["active"]:
+            session.pop("access_token", None)
+            return redirect(url_for("main.login"))
+
+        keycloak_ayr_user_group = current_app.config["KEYCLOAK_AYR_USER_GROUP"]
+        if not _check_if_user_has_access_to_ayr(
+            keycloak_ayr_user_group, decoded_token
+        ):
+            flash(
+                "TNA User is logged in but does not have access to AYR. Please contact your admin."
+            )
+            return redirect(url_for("main.index"))
+
+        flash("TNA User is logged in and has access to AYR.")
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+def _check_if_user_has_access_to_ayr(keycloak_ayr_user_group, decoded_token):
+    groups = decoded_token["groups"]
+    group_exists = False
+    for group in groups:
+        if keycloak_ayr_user_group in group:
+            group_exists = True
+    return group_exists

app/main/aws/open_search.py

@@ -1,29 +1,22 @@
 from typing import Tuple
 
 import boto3
+from flask import current_app
 from opensearchpy import AWSV4SignerAuth, OpenSearch
 
-from app.main.aws.parameter import (
-    get_aws_environment_prefix,
-    get_parameter_store_key_value,
-)
 
+def generate_open_search_client_from_current_app_config() -> OpenSearch:
+    """
+    Generate an OpenSearch client with the specified configuration for the AYR application.
 
-def get_open_search_index_from_aws_params() -> str:
-    return get_parameter_store_key_value(
-        get_aws_environment_prefix() + "AWS_OPEN_SEARCH_INDEX"
-    )
-
-
-def generate_open_search_client_from_aws_params() -> OpenSearch:
-    host = get_parameter_store_key_value(
-        get_aws_environment_prefix() + "AWS_OPEN_SEARCH_HOST"
-    )
-    http_auth = _get_open_search_http_auth()
-
+    Returns:
+        OpenSearch: An OpenSearch client configured with settings obtained from the current app's configuration.
+    """
     open_search_client = OpenSearch(
-        hosts=[{"host": host, "port": 443}],
-        http_auth=http_auth,
+        hosts=[
+            {"host": current_app.config["AWS_OPEN_SEARCH_HOST"], "port": 443}
+        ],
+        http_auth=get_open_search_http_auth(),
         use_ssl=True,
         verify_certs=True,
         http_compress=True,
@@ -33,29 +26,26 @@ def generate_open_search_client_from_aws_params() -> OpenSearch:
     return open_search_client
 
 
-def _get_open_search_http_auth(
-    auth_method: str = "username_password",
-) -> Tuple[str, str] | AWSV4SignerAuth:
-    if auth_method == "username_password":
-        return _get_open_search_username_password_auth()
-    return _get_open_search_iam_auth()
+def get_open_search_http_auth(iam=False) -> Tuple[str, str] | AWSV4SignerAuth:
+    """
+    Get the authentication method for OpenSearch.
 
+    Args:
+        iam (bool): A boolean indicating whether IAM authentication should be used.
 
-def _get_open_search_username_password_auth() -> Tuple[str, str]:
-    username = get_parameter_store_key_value(
-        get_aws_environment_prefix() + "AWS_OPEN_SEARCH_USERNAME"
-    )
-    password = get_parameter_store_key_value(
-        get_aws_environment_prefix() + "AWS_OPEN_SEARCH_PASSWORD"
-    )
-    return (username, password)
+    Returns:
+        Tuple[str, str] | AWSV4SignerAuth: Depending on the IAM parameter
+    """
+    if not iam:
+        return (
+            current_app.config["AWS_OPEN_SEARCH_USERNAME"],
+            current_app.config["AWS_OPEN_SEARCH_PASSWORD"],
+        )
+    return _get_open_search_iam_auth(current_app.config["AWS_REGION"])
 
 
-def _get_open_search_iam_auth() -> AWSV4SignerAuth:
+def _get_open_search_iam_auth(aws_region) -> AWSV4SignerAuth:
     credentials = boto3.Session().get_credentials()
-    aws_region = get_parameter_store_key_value(
-        get_aws_environment_prefix() + "AWS_REGION"
-    )
     service = "es"
     aws_auth = AWSV4SignerAuth(credentials, aws_region, service)
     return aws_auth

app/main/aws/parameter.py

@@ -6,7 +6,6 @@
 
 def get_aws_environment_prefix() -> str:
     environment_name = get_parameter_store_key_value("ENVIRONMENT_NAME")
-    print(environment_name)
     return "/" + environment_name + "/"
 
 

app/main/routes.py

@@ -1,37 +1,27 @@
-import os
-
+import keycloak
 from flask import (
+    current_app,
     flash,
     json,
     make_response,
     redirect,
     render_template,
     request,
     session,
+    url_for,
 )
 from flask_wtf.csrf import CSRFError
-from keycloak import KeycloakOpenID
 from werkzeug.exceptions import HTTPException
 
 from app.main import bp
+from app.main.authorize.keycloak_login_required_decorator import (
+    access_token_login_required,
+)
 from app.main.forms import CookiesForm
 from app.main.search import search_logic
 
 from .forms import SearchForm
 
-KEYCLOAK_BASE_URI = os.getenv("KEYCLOAK_BASE_URI")
-KEYCLOAK_CLIENT_ID = os.getenv("KEYCLOAK_CLIENT_ID")
-KEYCLOAK_REALM_NAME = os.getenv("KEYCLOAK_REALM_NAME")
-KEYCLOAK_CLIENT_SECRET = os.getenv("KEYCLOAK_CLIENT_SECRET")
-
-# Configure client
-keycloak_openid = KeycloakOpenID(
-    server_url=KEYCLOAK_BASE_URI,
-    client_id=KEYCLOAK_CLIENT_ID,
-    realm_name=KEYCLOAK_REALM_NAME,
-    client_secret_key=KEYCLOAK_CLIENT_SECRET,
-)
-
 
 @bp.route("/", methods=["GET"])
 def index():
@@ -40,9 +30,14 @@ def index():
 
 @bp.route("/login", methods=["GET"])
 def login():
-    # Get Code With Oauth Authorization Request
+    keycloak_openid = keycloak.KeycloakOpenID(
+        server_url=current_app.config["KEYCLOAK_BASE_URI"],
+        client_id=current_app.config["KEYCLOAK_CLIENT_ID"],
+        realm_name=current_app.config["KEYCLOAK_REALM_NAME"],
+        client_secret_key=current_app.config["KEYCLOAK_CLIENT_SECRET"],
+    )
     auth_url = keycloak_openid.auth_url(
-        redirect_uri="http://localhost:5000/callback",
+        redirect_uri=f"{current_app.config['APP_BASE_URL']}/callback",
         scope="email",
         state="your_state_info",
     )
@@ -53,11 +48,16 @@ def login():
 @bp.route("/callback", methods=["GET"])
 def callback():
     code = request.args.get("code")
-
+    keycloak_openid = keycloak.KeycloakOpenID(
+        server_url=current_app.config["KEYCLOAK_BASE_URI"],
+        client_id=current_app.config["KEYCLOAK_CLIENT_ID"],
+        realm_name=current_app.config["KEYCLOAK_REALM_NAME"],
+        client_secret_key=current_app.config["KEYCLOAK_CLIENT_SECRET"],
+    )
     access_token_response = keycloak_openid.token(
         grant_type="authorization_code",
         code=code,
-        redirect_uri="http://localhost:5000/callback",
+        redirect_uri=f"{current_app.config['APP_BASE_URL']}/callback",
     )
 
     session["access_token_response"] = access_token_response
@@ -67,6 +67,8 @@ def callback():
     session["token_scope"] = access_token_response["scope"]
     session["session_state"] = access_token_response["session_state"]
 
+    return redirect(url_for("main.poc_search"))
+
 
 @bp.route("/accessibility", methods=["GET"])
 def accessibility():
@@ -89,14 +91,17 @@ def results():
 
 
 @bp.route("/poc-search-view", methods=["POST", "GET"])
+@access_token_login_required
 def poc_search():
     form = SearchForm()
     results = []
     query = request.form.get("query", "").lower()
 
     if query:
         open_search_response = (
-            search_logic.generate_open_search_client_and_make_poc_search(query)
+            search_logic.generate_open_search_client_and_make_poc_search(
+                query, current_app.config["AWS_OPEN_SEARCH_INDEX"]
+            )
         )
         results = open_search_response["hits"]["hits"]
         session["search_results"] = results
@@ -112,6 +117,7 @@ def poc_search():
 
 
 @bp.route("/record", methods=["GET"])
+@access_token_login_required
 def record():
     """
     Render the record details page.

app/main/search/search_logic.py

@@ -4,12 +4,20 @@
 from opensearchpy import ImproperlyConfigured
 
 from app.main.aws.open_search import (
-    generate_open_search_client_from_aws_params,
-    get_open_search_index_from_aws_params,
+    generate_open_search_client_from_current_app_config,
 )
 
 
-def generate_open_search_client_and_make_poc_search(query: str) -> Any:
+def generate_open_search_client_and_make_poc_search(query: str, index) -> Any:
+    open_search_client = generate_open_search_client_from_current_app_config()
+    try:
+        open_search_client.ping()
+    except ImproperlyConfigured as e:
+        logging.error("OpenSearch client improperly configured: " + str(e))
+        raise e
+
+    logging.info("OpenSearch client has been connected successfully")
+
     fields = [
         "legal_status",
         "description",
@@ -22,17 +30,7 @@ def generate_open_search_client_and_make_poc_search(query: str) -> Any:
         "Consignment_Series",
         "Contact_Name",
     ]
-    open_search_client = generate_open_search_client_from_aws_params()
 
-    try:
-        open_search_client.ping()
-    except ImproperlyConfigured as e:
-        logging.error("OpenSearch client improperly configured: " + str(e))
-        raise e
-
-    logging.info("OpenSearch client has been connected successfully")
-
-    open_search_index = get_open_search_index_from_aws_params()
     open_search_query = {
         "query": {
             "multi_match": {
@@ -43,7 +41,8 @@ def generate_open_search_client_and_make_poc_search(query: str) -> Any:
             }
         }
     }
+
     search_results = open_search_client.search(
-        body=open_search_query, index=open_search_index
+        body=open_search_query, index=index
     )
     return search_results

app/templates/main/index.html

@@ -5,6 +5,7 @@
 {% endblock %}
 
 {% block content %}
+{{ super() }}
 <div class="govuk-grid-row">
   <div class="govuk-grid-column-two-thirds">
     <h1 class="govuk-heading-l">Hello, World!</h1>