test_auth.py 3.8 KB
Newer Older
ethiraj krishnamanaidu's avatar
ethiraj krishnamanaidu committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# Copyright 2021 Schlumberger
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Luc Yriarte's avatar
Luc Yriarte committed
15
16
import requests
import pytest
17
18
import datetime
import jwt
Luc Yriarte's avatar
Luc Yriarte committed
19
20
21
22
23
24
25
26
27

payload = {}

@pytest.fixture
def skip_if_gcp_environment(base_url):
    """
        In GCP environment there is no AuthorizationPolicy set. Certain tests may fail on GCP
        and this fixture aims to skip a test case when targeted environment is GCP.
    """
28
    response = requests.request("GET", f"{base_url}/about", verify=False)
Luc Yriarte's avatar
Luc Yriarte committed
29
30
31
32
33
34
35
36
37
    assert response.status_code == 200
    about_response = response.json()

    if about_response.get("cloudEnvironment") == "gcp":
        pytest.skip('skipped on this cloud provider because no AuthorizationPolicy in place')


# Test for expired token
def test_expired_token_returns_40X(base_url, check_cert, token):
38
    url = f"{base_url}/about"
39
    token_expired = jwt.encode({"email":"nobody@example.com", "exp":datetime.datetime.utcnow() - datetime.timedelta(seconds=300)}, key="secret", algorithm="HS256")
Luc Yriarte's avatar
Luc Yriarte committed
40
    headers = {
Luc Yriarte's avatar
Luc Yriarte committed
41
        'Authorization': f"Bearer {token_expired}"
Luc Yriarte's avatar
Luc Yriarte committed
42
43
44
45
46
47
48
49
50
51
52
53
54
55
    }
    response = requests.request("GET", url, headers=headers, data=payload, verify=check_cert)
    assert response.status_code == 401
    
# Test for no token on some paths where JWT token is NOT required due to the AuthorizationPolicy. Test to ensure headers are present for docs endpoint
def test_notoken_paths_returns_20X_docs(base_url, check_cert, token):
    
    url = f"{base_url}/docs"
    headers = {}
    response = requests.request("GET", url, headers=headers, data=payload, verify=check_cert)
    assert response.status_code == 200
    assert 'content-security-policy' in response.headers

# Test for no token on some paths where JWT token is NOT required due to the AuthorizationPolicy
56
@pytest.mark.parametrize("path", ["docs", "openapi.json", "about"])
Luc Yriarte's avatar
Luc Yriarte committed
57
58
59
60
61
62
63
64
65
66
67
def test_notoken_paths_returns_20X(base_url, check_cert, token, path):

    url = f"{base_url}/{path}"
    headers = {}
    response = requests.request("GET", url, headers=headers, data=payload, verify=check_cert)
    assert response.status_code == 200

# Test for no token on some paths where JWT token is required due to the AuthorizationPolicy
@pytest.mark.parametrize("path", ["version", "nonExistingPath"])
def test_notoken_returns_40X(base_url, check_cert, token, skip_if_gcp_environment, path):

68
    url = f"{base_url}/{path}"
Luc Yriarte's avatar
Luc Yriarte committed
69
70
71
72
73
74
75
76
    headers = {}
    response = requests.request("GET", url, headers=headers, data=payload, verify=check_cert)
    assert response.status_code == 403
    assert "access denied" in response.text


# Test for invalid token
def test_invalid_token_returns_40X(base_url, check_cert, token):
77
    url = f"{base_url}/about"
Luc Yriarte's avatar
Luc Yriarte committed
78
79
80
    blank = {}
    token_invalid = token[0:len(token) - 10]
    headers = {
81
        'Authorization': f"Bearer {token_invalid}"
Luc Yriarte's avatar
Luc Yriarte committed
82
83
84
85
86
87
88
89
    }

    response = requests.request("GET", url, headers=headers, data=blank, verify=check_cert)
    assert response.status_code == 401


# Test for unauthorized issuer
def test_invalid_issuer_token_returns_40X(base_url, check_cert, token):
90
    url = f"{base_url}/about"
Luc Yriarte's avatar
Luc Yriarte committed
91
    blank = {}
92
    token_no_iss = jwt.encode({"email": "nobody@example.com"}, key="secret", algorithm="HS256")
Luc Yriarte's avatar
Luc Yriarte committed
93
    headers = {
Luc Yriarte's avatar
Luc Yriarte committed
94
        'Authorization': f"Bearer {token_no_iss}"
Luc Yriarte's avatar
Luc Yriarte committed
95
96
97
    }
    response = requests.request("GET", url, headers=headers, data=blank, verify=check_cert)
    assert response.status_code == 401