Merge pull request #31 from TheJacksonLaboratory/G3-187-core-security…

…-test core/security test coverage > 80%
TheJacksonLaboratory · Feb 23, 2024 · 61af93a · 61af93a
2 parents 2eeb080 + cb363a4
commit 61af93a
Show file tree

Hide file tree

Showing 4 changed files with 337 additions and 1 deletion.
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "geneweaver-api"
-version = "0.2.0"
+version = "0.2.1a1"
 description = "The Geneweaver API"
 authors = ["Jax Computational Sciences <cssc@jax.org>"]
 readme = "README.md"

diff --git a/tests/core/test_security.py b/tests/core/test_security.py
@@ -0,0 +1,297 @@
+"""Tests for core security."""
+
+from unittest.mock import patch
+
+import pytest
+from fastapi import HTTPException
+from fastapi.security import HTTPAuthorizationCredentials, SecurityScopes
+from geneweaver.api.core.exceptions import (
+    Auth0UnauthenticatedException,
+    Auth0UnauthorizedException,
+)
+from geneweaver.api.core.security import Auth0, UserInternal
+from jose import jwt
+
+from tests.data import test_jwt_keys_data
+
+private_key = test_jwt_keys_data.get("test_private_key")
+public_key = test_jwt_keys_data.get("test_public_key")
+
+test_audience = "https://gw.test.org"
+test_domain = "gw.test.auth0.com"
+test_email = "test@test.org"
+test_name = "Test Name"
+
+
+# custom class to be the mock return value
+# will override the requests.Response returned from requests.get
+class MockGetResponse:
+    """Mock for get response."""
+
+    # mock json()
+    @staticmethod
+    def json() -> dict:
+        """Json response."""
+        return {"mock_key": "mock_response"}
+
+
+def do_auth():
+    """Initialize Auth object with test config."""
+    auth = Auth0(
+        domain=test_domain,
+        api_audience=test_audience,
+        scopes={
+            "openid profile email": "read",
+        },
+        auto_error=True,
+        email_auto_error=True,
+    )
+
+    auth.jwks = public_key
+
+    return auth
+
+
+@patch("geneweaver.api.core.security.requests")
+def create_test_token(mock_requests, claims=None):
+    """Create a valid RS256 test JWT token."""
+    mock_requests.get.return_value = MockGetResponse()
+
+    # claims
+    if claims is None:
+        to_encode = {
+            f"{test_audience}/claims/email": test_email,
+            "iss": f"https://{test_domain}/",
+            "aud": test_audience,
+            "name": test_name,
+            "scope": "openid profile email",
+        }
+    else:
+        to_encode = claims
+
+    token = jwt.encode(to_encode, private_key, algorithm="RS256")
+
+    return token
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.SecurityScopes")
+@patch("geneweaver.api.core.security.requests")
+async def test_get_user_no_creds_http_error(mock_requests, mock_security_scope):
+    """Test get user with no credetials in the request."""
+    auth = do_auth()
+
+    with pytest.raises(expected_exception=HTTPException):
+        await auth.get_user_strict(security_scopes=mock_security_scope, creds=None)
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.SecurityScopes")
+@patch("geneweaver.api.core.security.requests")
+async def test_invalid_token_format(mock_requests, mock_security_scope):
+    """Test invalid token in credentials."""
+    auth = do_auth()
+
+    creds = HTTPAuthorizationCredentials(scheme="", credentials="token")
+
+    with pytest.raises(expected_exception=Auth0UnauthenticatedException):
+        await auth.get_user(
+            security_scopes=mock_security_scope,
+            creds=creds,
+            auto_error_auth=True,
+            disallow_public=True,
+        )
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.SecurityScopes")
+@patch("geneweaver.api.core.security.jwt.get_unverified_header")
+@patch("geneweaver.api.core.security.requests")
+async def test_valid_jwt_token(
+    mock_requests, mock_jwt_unverified_header, mock_security_scope
+):
+    """Test get user with no credetials in the request."""
+    auth = do_auth()
+    mock_jwt_unverified_header.return_value = private_key
+
+    # get test token
+    token = create_test_token()
+    creds = HTTPAuthorizationCredentials(credentials=token, scheme="")
+
+    user: UserInternal = await auth.get_user(
+        security_scopes=mock_security_scope,
+        creds=creds,
+        auto_error_auth=True,
+        disallow_public=False,
+    )
+
+    print(user)
+    assert user is not None
+    assert user.email == test_email
+    assert user.name == test_name
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.SecurityScopes")
+@patch("geneweaver.api.core.security.jwt.get_unverified_header")
+@patch("geneweaver.api.core.security.requests")
+async def test_get_user_strict_valid_jwt_token(
+    mock_requests, mock_jwt_unverified_header, mock_security_scope
+):
+    """Test get user strict with a valid token."""
+    auth = do_auth()
+    mock_jwt_unverified_header.return_value = private_key
+
+    # get test token
+    token = create_test_token()
+    creds = HTTPAuthorizationCredentials(credentials=token, scheme="")
+
+    user: UserInternal = await auth.get_user_strict(
+        security_scopes=mock_security_scope, creds=creds
+    )
+
+    print(user)
+    assert user is not None
+    assert user.email == test_email
+    assert user.name == test_name
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.jwt.get_unverified_header")
+@patch("geneweaver.api.core.security.requests")
+async def test_get_user_with_scopes(mock_requests, mock_jwt_unverified_header):
+    """Test get user with secuirty scopes."""
+    auth = do_auth()
+    mock_jwt_unverified_header.return_value = private_key
+
+    # get test token
+    token = create_test_token()
+    creds = HTTPAuthorizationCredentials(credentials=token, scheme="")
+
+    scopes = SecurityScopes(scopes=["openid", "profile", "email"])
+    user: UserInternal = await auth.get_user_strict(security_scopes=scopes, creds=creds)
+
+    print(user)
+    assert user is not None
+    assert user.email == test_email
+    assert user.name == test_name
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.jwt.get_unverified_header")
+@patch("geneweaver.api.core.security.requests")
+async def test_authenticated(mock_requests, mock_jwt_unverified_header):
+    """Test get user authenticated."""
+    auth = do_auth()
+    mock_jwt_unverified_header.return_value = private_key
+
+    # get test token
+    token = create_test_token()
+    creds = HTTPAuthorizationCredentials(credentials=token, scheme="")
+
+    scopes = SecurityScopes(scopes=["openid", "profile", "email"])
+    authenticated = await auth.authenticated(security_scopes=scopes, creds=creds)
+
+    assert authenticated is True
+
+    token = "token"
+    creds = HTTPAuthorizationCredentials(credentials=token, scheme="")
+    authenticated = await auth.authenticated(security_scopes=scopes, creds=creds)
+
+    assert authenticated is False
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.jwt.get_unverified_header")
+@patch("geneweaver.api.core.security.requests")
+async def test_is_user_public(mock_requests, mock_jwt_unverified_header):
+    """Test is user public."""
+    auth = do_auth()
+    mock_jwt_unverified_header.return_value = private_key
+
+    # get test token
+    token = create_test_token()
+    creds = HTTPAuthorizationCredentials(credentials=token, scheme="")
+
+    scopes = SecurityScopes(scopes=["openid", "profile", "email"])
+    authenticated = await auth.public(security_scopes=scopes, creds=creds)
+
+    assert authenticated is False
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.SecurityScopes")
+@patch("geneweaver.api.core.security.jwt.get_unverified_header")
+@patch("geneweaver.api.core.security.requests")
+async def test_is_user_not_public(
+    mock_requests, mock_jwt_unverified_header, mock_security_scope
+):
+    """Test user is not public."""
+    auth = do_auth()
+    is_public = await auth.get_user(
+        security_scopes=mock_security_scope, creds=None, disallow_public=False
+    )
+
+    assert is_public is None
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.SecurityScopes")
+@patch("geneweaver.api.core.security.jwt.get_unverified_header")
+@patch("geneweaver.api.core.security.requests")
+async def test_invalid_claim(
+    mock_requests, mock_jwt_unverified_header, mock_security_scope
+):
+    """Test get user exception with invalid claim."""
+    auth = do_auth()
+    mock_jwt_unverified_header.return_value = private_key
+
+    to_encode = {
+        f"{test_audience}/claims/email": test_email,
+        "name": test_name,
+        "scope": "openid profile email",
+    }
+
+    # get test token
+    token = create_test_token(claims=to_encode)
+    creds = HTTPAuthorizationCredentials(credentials=token, scheme="")
+
+    with pytest.raises(expected_exception=Auth0UnauthenticatedException):
+        await auth.get_user(
+            security_scopes=mock_security_scope,
+            creds=creds,
+            auto_error_auth=True,
+            disallow_public=True,
+        )
+
+
+@pytest.mark.asyncio()
+@patch("geneweaver.api.core.security.SecurityScopes")
+@patch("geneweaver.api.core.security.jwt.get_unverified_header")
+@patch("geneweaver.api.core.security.requests")
+async def test_missing_claim_email_error_claim(
+    mock_requests, mock_jwt_unverified_header, mock_security_scope
+):
+    """Test get user exception with missing email in claim."""
+    auth = do_auth()
+    mock_jwt_unverified_header.return_value = private_key
+
+    to_encode = {
+        f"{test_audience}/claims/email": None,
+        "iss": f"https://{test_domain}/",
+        "aud": test_audience,
+        "name": test_name,
+        "scope": "openid profile email",
+    }
+
+    # get test token
+    token = create_test_token(claims=to_encode)
+    creds = HTTPAuthorizationCredentials(credentials=token, scheme="")
+
+    with pytest.raises(expected_exception=Auth0UnauthorizedException):
+        await auth.get_user(
+            security_scopes=mock_security_scope,
+            creds=creds,
+            auto_error_auth=True,
+            disallow_public=True,
+        )
diff --git a/tests/data/__init__.py b/tests/data/__init__.py
@@ -18,6 +18,11 @@
 
 publications_json = importlib.resources.read_text("tests.data", "publications.json")
 
+jwt_test_keys_json = importlib.resources.read_text(
+    "tests.data", "security_jwt_RS256_keys.json"
+)
+
+
 ## laod and returns JSON string as a dictionary
 
 # geneset test data
@@ -58,3 +63,9 @@
         "publication_by_pubmed_id"
     ),
 }
+
+# Json web token keys data =
+test_jwt_keys_data = {
+    "test_private_key": json.loads(jwt_test_keys_json).get("private_key"),
+    "test_public_key": json.loads(jwt_test_keys_json).get("public_key"),
+}
diff --git a/tests/data/security_jwt_RS256_keys.json b/tests/data/security_jwt_RS256_keys.json
@@ -0,0 +1,28 @@
+{
+  "private_key": {
+    "p": "8MfRfDEJOJ9TkKmk_G5A757mE7XP2P8FvDbfw6onr3RzVjFMnsrv6M0CpOcYzQU_UEmhgxWrB2fOi3yUr2uEfBdhBF7mkmiV58dPZiEqPBU8HISh42kioppbGAtuAGUVAcqbLoFBqiEOldKwS-bG4xg082n4EIDIDlxg1BqV_zc",
+    "kty": "RSA",
+    "q": "uB6UMe9Wm1eTMYRuCmduqzhayQgvY_bgyzqym1qNte9nn9Bql0U95fvn_-VmOB7kY9-Hx-iJWBJWEr_fQnXQldQ5jOJVmIoyRpOl9pquDHUXRwkVkOrZUbLdBJiAVQx7It2yJDhwvi9XwPeVUYEnGKaAvNbebUrMFydGs5AQ-CU",
+    "d": "OVAo4Yan9WynANP7vSu19owW93-WORjxg3R5GHLJaxrX9nk_ztgIrgeSDad6bfmpPWp34W7deWp5Qmh47eMm-upcvbyUQFkOkZ62tujD-dwg_bj7XexhUoNWDEGnaiBcqsbF7BbPIKN5ckxgf9M_ZHzl7hmYcUYk3UlH8mUqlXw-wfOYu_6yH584zXTRtczXAABdFsiAipG7LGpx1zDZOutGxMW4VEIiZsAHj0p4_YndhlIL7zS3TAAsZE7X7X2fA0qYPgWwluke6hATTt17t6XkuqBCgmUMUmSE_haKwtZ6ZZPJiyT9LP7_gbMAEJvwmXrDqB1hi5nQ395kTWpccQ",
+    "e": "AQAB",
+    "use": "sig",
+    "kid": "I3upxxbKLxEJdooVxuuLsTvu7sqJeMtBAVXHwvTs-uQ",
+    "qi": "YAHE9VoRO7iKG2gHioPQ-25URveYTf8G-K2DJLsXphwDPPppGqIIpP7lMTJwC9ZD1YOFfVMwer_Hd9Gjj93WNw-Rmytfp3DIXuu0o1kWZyrDfyyJAW8nakUtkJKROt_vaBIkV_lraR02UAdkPclO6Sc4dvN_ZH1ApsuMgvuUXDs",
+    "dp": "3DmG6xZWnslrPzdKxe95yTEGsyRp1Ml8T2fJRkdNQPc7vqwcrmhjAgTw1C7iyjJwdFjENwcMhRt3GLF7tO6cIHupqru6HFM4OORdRMY0wPuTHWpaP4ubuCmCA_4AQLAzhI3xXZmvm5Hcq0AnK2UKqA8t7y0PTNjdIfVwQs-GPgU",
+    "alg": "RS256",
+    "dq": "Qifan8abm91vqg8nat2XSjZJiIpEXOrMArnoiyGSYZjP5wCADDJ49zX4Ol42yFtxPOGIbDAFiXutKbd_hOXIOM20kAaTMugVAH701xLlDtzTrFZ7RULdKxnViF0zX1vIstJtu8371Jo2McPEBzEc1yKchz29Vg_WHUujf8l4D3E",
+    "n": "rSxhXkxDRrSJ63-kDBRWbBbqQS5PdrOStkG5SgxwPf2g4tCCZaSMGTaluVf6Zio-ixRK0-n8pr9gZBtr7Yqk-oQLVOyMjqOSgLYkj2uCWypDfGZdmxuV86N5QYBWoQh1x9AKzD36eEa_CQ_3dQ5Rt_W8kkDNlthYOuWlurU0yg3v-l8aNeVLRQcCKJHfO3SBqfJ7ViDwR4fwfTpxu66IKHJWn7xSrDEw7DYvPHTrma47hBuOwU96UyoTWRPBfkFdkAi0VvN-fqnwyG2Zt60Rn8E6oMceP0Dj78B_Duep2VjHlQrRBmWfgND-cYugI7vcIWt2jG82x0BaaP7NClsq8w"
+  },
+  "public_key": {
+    "keys": [
+      {
+        "kty": "RSA",
+        "e": "AQAB",
+        "use": "sig",
+        "kid": "I3upxxbKLxEJdooVxuuLsTvu7sqJeMtBAVXHwvTs-uQ",
+        "alg": "RS256",
+        "n": "rSxhXkxDRrSJ63-kDBRWbBbqQS5PdrOStkG5SgxwPf2g4tCCZaSMGTaluVf6Zio-ixRK0-n8pr9gZBtr7Yqk-oQLVOyMjqOSgLYkj2uCWypDfGZdmxuV86N5QYBWoQh1x9AKzD36eEa_CQ_3dQ5Rt_W8kkDNlthYOuWlurU0yg3v-l8aNeVLRQcCKJHfO3SBqfJ7ViDwR4fwfTpxu66IKHJWn7xSrDEw7DYvPHTrma47hBuOwU96UyoTWRPBfkFdkAi0VvN-fqnwyG2Zt60Rn8E6oMceP0Dj78B_Duep2VjHlQrRBmWfgND-cYugI7vcIWt2jG82x0BaaP7NClsq8w"
+      }
+    ]
+  }
+}