mirror of
https://github.com/GSA/notifications-api.git
synced 2026-09-06 11:08:24 -04:00
Added the before_request so that all calls must have a valid token.
Next is to get all the rest tests to pass again.
This commit is contained in:
@@ -46,6 +46,13 @@ def init_app(app):
|
|||||||
if key in os.environ:
|
if key in os.environ:
|
||||||
app.config[key] = convert_to_boolean(os.environ[key])
|
app.config[key] = convert_to_boolean(os.environ[key])
|
||||||
|
|
||||||
|
@app.before_request
|
||||||
|
def required_authentication():
|
||||||
|
from app.authentication import auth
|
||||||
|
error = auth.requires_auth()
|
||||||
|
if error:
|
||||||
|
return error
|
||||||
|
|
||||||
|
|
||||||
def convert_to_boolean(value):
|
def convert_to_boolean(value):
|
||||||
"""Turn strings to bools if they look like them
|
"""Turn strings to bools if they look like them
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ from flask import Blueprint
|
|||||||
status = Blueprint('status', __name__)
|
status = Blueprint('status', __name__)
|
||||||
|
|
||||||
|
|
||||||
@status.route('/_status')
|
@status.route('/_status', methods=['GET', 'POST'])
|
||||||
def show_status():
|
def show_status():
|
||||||
return jsonify(
|
return jsonify(
|
||||||
status="ok",
|
status="ok",
|
||||||
|
|||||||
0
tests/app/authentication/__init__.py
Normal file
0
tests/app/authentication/__init__.py
Normal file
122
tests/app/authentication/test_authentication.py
Normal file
122
tests/app/authentication/test_authentication.py
Normal file
@@ -0,0 +1,122 @@
|
|||||||
|
import pytest
|
||||||
|
from flask import json, url_for
|
||||||
|
from client.authentication import create_jwt_token
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_not_allow_request_with_no_token(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
response = client.get(url_for('status.show_status'))
|
||||||
|
assert response.status_code == 401
|
||||||
|
data = json.loads(response.get_data())
|
||||||
|
assert data['error'] == 'Unauthorized, authentication token must be provided'
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_not_allow_request_with_incorrect_header(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
response = client.get(url_for('status.show_status'),
|
||||||
|
headers={'Authorization': 'Basic 1234'})
|
||||||
|
assert response.status_code == 401
|
||||||
|
data = json.loads(response.get_data())
|
||||||
|
assert data['error'] == 'Unauthorized, authentication bearer scheme must be used'
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_not_allow_request_with_incorrect_token(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
response = client.get(url_for('status.show_status'),
|
||||||
|
headers={'Authorization': 'Bearer 1234'})
|
||||||
|
assert response.status_code == 403
|
||||||
|
data = json.loads(response.get_data())
|
||||||
|
assert data['error'] == 'Invalid token: signature'
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_not_allow_incorrect_path(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
token = create_jwt_token(request_method="GET", request_path="/bad", secret="secret", client_id="client_id")
|
||||||
|
response = client.get(url_for('status.show_status'),
|
||||||
|
headers={'Authorization': "Bearer {}".format(token)})
|
||||||
|
assert response.status_code == 403
|
||||||
|
data = json.loads(response.get_data())
|
||||||
|
assert data['error'] == 'Invalid token: request'
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_not_allow_incorrect_method(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
token = create_jwt_token(request_method="POST", request_path="/", secret="secret", client_id="client_id")
|
||||||
|
response = client.get(url_for('status.show_status'),
|
||||||
|
headers={'Authorization': "Bearer {}".format(token)})
|
||||||
|
assert response.status_code == 403
|
||||||
|
data = json.loads(response.get_data())
|
||||||
|
assert data['error'] == 'Invalid token: request'
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_not_allow_invalid_secret(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
token = create_jwt_token(request_method="POST", request_path="/", secret="not-so-secret",
|
||||||
|
client_id="client_id")
|
||||||
|
response = client.get(url_for('status.show_status'),
|
||||||
|
headers={'Authorization': "Bearer {}".format(token)})
|
||||||
|
assert response.status_code == 403
|
||||||
|
data = json.loads(response.get_data())
|
||||||
|
assert data['error'] == 'Invalid token: signature'
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_allow_valid_token(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
token = create_jwt_token(request_method="GET",
|
||||||
|
request_path=url_for('status.show_status'),
|
||||||
|
secret="secret",
|
||||||
|
client_id="client_id")
|
||||||
|
response = client.get(url_for('status.show_status'),
|
||||||
|
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_allow_valid_token_with_post_body(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
json_body = json.dumps({
|
||||||
|
"key1": "value1",
|
||||||
|
"key2": "value2",
|
||||||
|
"key3": "value3"
|
||||||
|
})
|
||||||
|
token = create_jwt_token(
|
||||||
|
request_method="POST",
|
||||||
|
request_path=url_for('status.show_status'),
|
||||||
|
secret="secret",
|
||||||
|
client_id="client_id",
|
||||||
|
request_body=json_body
|
||||||
|
)
|
||||||
|
response = client.post(url_for('status.show_status'),
|
||||||
|
data=json_body,
|
||||||
|
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_not_allow_valid_token_with_invalid_post_body(notify_api):
|
||||||
|
with notify_api.test_request_context():
|
||||||
|
with notify_api.test_client() as client:
|
||||||
|
json_body = json.dumps({
|
||||||
|
"key1": "value1",
|
||||||
|
"key2": "value2",
|
||||||
|
"key3": "value3"
|
||||||
|
})
|
||||||
|
token = create_jwt_token(
|
||||||
|
request_method="POST",
|
||||||
|
request_path=url_for('status.show_status'),
|
||||||
|
secret="secret",
|
||||||
|
client_id="client_id",
|
||||||
|
request_body=json_body
|
||||||
|
)
|
||||||
|
response = client.post(url_for('status.show_status'),
|
||||||
|
data="spurious",
|
||||||
|
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||||
|
assert response.status_code == 403
|
||||||
|
data = json.loads(response.get_data())
|
||||||
|
assert data['error'] == 'Invalid token: payload'
|
||||||
@@ -1,141 +0,0 @@
|
|||||||
import pytest
|
|
||||||
from flask import json
|
|
||||||
from client.authentication import create_jwt_token
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_not_allow_request_with_no_token(notify_api):
|
|
||||||
response = notify_api.test_client().get("/")
|
|
||||||
assert response.status_code == 401
|
|
||||||
data = json.loads(response.get_data())
|
|
||||||
assert data['error'] == 'Unauthorized, authentication token must be provided'
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_not_allow_request_with_incorrect_header(notify_api):
|
|
||||||
response = notify_api.test_client().get(
|
|
||||||
"/",
|
|
||||||
headers={
|
|
||||||
'Authorization': 'Basic 1234'
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert response.status_code == 401
|
|
||||||
data = json.loads(response.get_data())
|
|
||||||
assert data['error'] == 'Unauthorized, authentication bearer scheme must be used'
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_not_allow_request_with_incorrect_token(notify_api):
|
|
||||||
response = notify_api.test_client().get(
|
|
||||||
"/",
|
|
||||||
headers={
|
|
||||||
'Authorization': 'Bearer 1234'
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert response.status_code == 403
|
|
||||||
data = json.loads(response.get_data())
|
|
||||||
assert data['error'] == 'Invalid token: signature'
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_not_allow_incorrect_path(notify_api):
|
|
||||||
token = create_jwt_token(request_method="GET", request_path="/bad", secret="secret", client_id="client_id")
|
|
||||||
response = notify_api.test_client().get(
|
|
||||||
"/",
|
|
||||||
headers={
|
|
||||||
'Authorization': "Bearer {}".format(token)
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert response.status_code == 403
|
|
||||||
data = json.loads(response.get_data())
|
|
||||||
assert data['error'] == 'Invalid token: request'
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_not_allow_incorrect_method(notify_api):
|
|
||||||
token = create_jwt_token(request_method="POST", request_path="/", secret="secret", client_id="client_id")
|
|
||||||
response = notify_api.test_client().get(
|
|
||||||
"/",
|
|
||||||
headers={
|
|
||||||
'Authorization': "Bearer {}".format(token)
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert response.status_code == 403
|
|
||||||
data = json.loads(response.get_data())
|
|
||||||
assert data['error'] == 'Invalid token: request'
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_not_allow_invalid_secret(notify_api):
|
|
||||||
token = create_jwt_token(request_method="POST", request_path="/", secret="not-so-secret", client_id="client_id")
|
|
||||||
response = notify_api.test_client().get(
|
|
||||||
"/",
|
|
||||||
headers={
|
|
||||||
'Authorization': "Bearer {}".format(token)
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert response.status_code == 403
|
|
||||||
data = json.loads(response.get_data())
|
|
||||||
assert data['error'] == 'Invalid token: signature'
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_allow_valid_token(notify_api):
|
|
||||||
token = create_jwt_token(request_method="GET", request_path="/", secret="secret", client_id="client_id")
|
|
||||||
response = notify_api.test_client().get(
|
|
||||||
"/",
|
|
||||||
headers={
|
|
||||||
'Authorization': 'Bearer {}'.format(token)
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert response.status_code == 200
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_allow_valid_token_with_post_body(notify_api):
|
|
||||||
json_body = json.dumps({
|
|
||||||
"key1": "value1",
|
|
||||||
"key2": "value2",
|
|
||||||
"key3": "value3"
|
|
||||||
})
|
|
||||||
token = create_jwt_token(
|
|
||||||
request_method="POST",
|
|
||||||
request_path="/",
|
|
||||||
secret="secret",
|
|
||||||
client_id="client_id",
|
|
||||||
request_body=json_body
|
|
||||||
)
|
|
||||||
response = notify_api.test_client().post(
|
|
||||||
"/",
|
|
||||||
data=json_body,
|
|
||||||
headers={
|
|
||||||
'Authorization': 'Bearer {}'.format(token)
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert response.status_code == 200
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(reason="Authentication to be added.")
|
|
||||||
def test_should_not_allow_valid_token_with_invalid_post_body(notify_api):
|
|
||||||
json_body = json.dumps({
|
|
||||||
"key1": "value1",
|
|
||||||
"key2": "value2",
|
|
||||||
"key3": "value3"
|
|
||||||
})
|
|
||||||
token = create_jwt_token(
|
|
||||||
request_method="POST",
|
|
||||||
request_path="/",
|
|
||||||
secret="secret",
|
|
||||||
client_id="client_id",
|
|
||||||
request_body=json_body
|
|
||||||
)
|
|
||||||
response = notify_api.test_client().post(
|
|
||||||
"/",
|
|
||||||
data="spurious",
|
|
||||||
headers={
|
|
||||||
'Authorization': 'Bearer {}'.format(token)
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert response.status_code == 403
|
|
||||||
data = json.loads(response.get_data())
|
|
||||||
assert data['error'] == 'Invalid token: payload'
|
|
||||||
Reference in New Issue
Block a user