diff --git a/api/app.py b/api/app.py index 9ad9f6e533..ef43fe1ebc 100644 --- a/api/app.py +++ b/api/app.py @@ -3,6 +3,9 @@ from flask import Flask from flask_graphql import GraphQLView from waitress import serve + +from backend.security_check import SecurityAnalysisBackend + from db import ( db, DB_NAME, @@ -15,19 +18,21 @@ app = Flask(__name__) -app.config[ - 'SQLALCHEMY_DATABASE_URI'] = f'postgresql+psycopg2://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}' +app.config['SQLALCHEMY_DATABASE_URI'] = f'postgresql+psycopg2://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}' app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True app.debug = True db.init_app(app) +backend = SecurityAnalysisBackend(max_depth=10, max_cost=1000) + app.add_url_rule( '/graphql', view_func=GraphQLView.as_view( 'graphql', schema=schema, + backend=backend, graphiql=True ) ) diff --git a/api/backend/__init__.py b/api/backend/__init__.py new file mode 100644 index 0000000000..0524dc9484 --- /dev/null +++ b/api/backend/__init__.py @@ -0,0 +1,24 @@ +from typing import ( + Dict, + List +) +from graphql.language.ast import ( + FragmentDefinition, + OperationDefinition +) + + +def get_fragments(definitions) -> Dict[str, FragmentDefinition]: + return { + definition.name.value: definition + for definition in definitions + if isinstance(definition, FragmentDefinition) + } + + +def get_queries_and_mutations(definitions) -> List[OperationDefinition]: + return [ + definition + for definition in definitions + if isinstance(definition, OperationDefinition) + ] diff --git a/api/backend/cost_check.py b/api/backend/cost_check.py new file mode 100644 index 0000000000..8b4eddda57 --- /dev/null +++ b/api/backend/cost_check.py @@ -0,0 +1,72 @@ +from typing import ( + Dict +) +from graphql.language.ast import ( + Document, + FragmentDefinition, + OperationDefinition, + Node, + FragmentSpread, + Field, + InlineFragment +) +from backend import ( + get_fragments, + get_queries_and_mutations +) + +from backend.cost_map import cost_map + + +class CostLimitReached(Exception): + pass + + +def measure_cost(node: Node, fragments: Dict[str, FragmentDefinition]) -> int: + """ + A function which recursively measures the cost of a Graphene Query + :type node: Node + :param node: Graphql-core object used for query traversal/indexing + :type fragments: dict + :param fragments: The fragments of the query + :rtype: int + :return: The cost of the node + """ + if isinstance(node, FragmentSpread): + fragment = fragments.get(node.name.value) + return measure_cost(node=fragment, fragments=fragments) + + elif isinstance(node, Field): + if node.name.value.lower() in ["__schema", "__introspection"]: + return 0 + if not node.selection_set: + return cost_map.get(node.name.value, 1) + costs = [] + for selection in node.selection_set.selections: + cost = measure_cost(node=selection, fragments=fragments) + costs.append(cost) + return sum(costs) + cost_map.get(node.name.value, 1) + elif ( + isinstance(node, FragmentDefinition) + or isinstance(node, OperationDefinition) + or isinstance(node, InlineFragment) + ): + costs = [] + for selection in node.selection_set.selections: + cost = measure_cost(node=selection, fragments=fragments) + costs.append(cost) + return sum(costs) + else: + raise Exception("Unknown node") + + +def check_cost_analysis(max_cost: int, document: Document): + fragments = get_fragments(document.definitions) + queries = get_queries_and_mutations(document.definitions) + + for query in queries: + total_cost = measure_cost(query, fragments) + if total_cost > max_cost: + raise CostLimitReached( + 'Query cost is too high' + ) diff --git a/api/backend/cost_map.py b/api/backend/cost_map.py new file mode 100644 index 0000000000..e42b7593b3 --- /dev/null +++ b/api/backend/cost_map.py @@ -0,0 +1,5 @@ +cost_map = { + 'getSectorById': 1, + 'getOrgById': 1, + 'getGroupById': 1 +} diff --git a/api/backend/depth_check.py b/api/backend/depth_check.py new file mode 100644 index 0000000000..63ba6f566d --- /dev/null +++ b/api/backend/depth_check.py @@ -0,0 +1,69 @@ +from typing import Dict +from graphql.language.ast import ( + Document, + FragmentDefinition, + OperationDefinition, + Node, + FragmentSpread, + Field, + InlineFragment +) + +from backend import ( + get_fragments, + get_queries_and_mutations +) + + +class DepthLimitReached(Exception): + pass + + +def measure_depth(node: Node, fragments: Dict[str, FragmentDefinition]) -> int: + """ + A function which recursively measures the depth of a Graphene Query + :type node: Node + :param node: Graphql-core object used for query traversal/indexing + :type fragments: dict + :param fragments: The fragments of the query + :rtype: int + :return: The max depth of the node + """ + if isinstance(node, FragmentSpread): + fragment = fragments.get(node.name.value) + return measure_depth(node=fragment, fragments=fragments) + + elif isinstance(node, Field): + if node.name.value.lower() in ["__schema", "__introspection"]: + return 0 + if not node.selection_set: + return 1 + depths = [] + for selection in node.selection_set.selections: + depth = measure_depth(node=selection, fragments=fragments) + depths.append(depth) + return 1 + max(depths) + elif ( + isinstance(node, FragmentDefinition) + or isinstance(node, OperationDefinition) + or isinstance(node, InlineFragment) + ): + depths = [] + for selection in node.selection_set.selections: + depth = measure_depth(node=selection, fragments=fragments) + depths.append(depth) + return max(depths) + else: + raise Exception("Unknown node") + + +def check_max_depth(max_depth: int, document: Document): + fragments = get_fragments(document.definitions) + queries = get_queries_and_mutations(document.definitions) + + for query in queries: + depth = measure_depth(query, fragments) + if depth > max_depth: + raise DepthLimitReached( + 'Query is too complex' + ) diff --git a/api/backend/security_check.py b/api/backend/security_check.py new file mode 100644 index 0000000000..c026605436 --- /dev/null +++ b/api/backend/security_check.py @@ -0,0 +1,24 @@ +from typing import ( + Union, + Optional, + Any +) +from graphql import GraphQLDocument, GraphQLSchema +from graphql.backend.core import GraphQLCoreBackend +from graphql.language.ast import Document + +from backend.depth_check import check_max_depth +from backend.cost_check import check_cost_analysis + + +class SecurityAnalysisBackend(GraphQLCoreBackend): + def __init__(self, max_depth=10, max_cost=1000, executor: Optional[Any] = None): + super().__init__(executor=executor) + self.max_depth = max_depth + self.max_cost = max_cost + + def document_from_string(self, schema: GraphQLSchema, document_string: Union[Document, str]) -> GraphQLDocument: + document = super().document_from_string(schema, document_string) + check_max_depth(max_depth=self.max_depth, document=document.document_ast) + check_cost_analysis(max_cost=self.max_cost, document=document.document_ast) + return document diff --git a/api/resolvers/sectors.py b/api/resolvers/sectors.py index 86c1fa70ab..621d743b3d 100644 --- a/api/resolvers/sectors.py +++ b/api/resolvers/sectors.py @@ -1,38 +1,41 @@ from graphql import GraphQLError from schemas.sectors import Sectors, SectorsModel from model_enums.sectors import SectorEnums +from manage import app # Resolvers def resolve_get_sector_by_id(self, info, **kwargs): - """Return a sector by its row ID""" - sector_id = kwargs.get('id', 1) - query = Sectors.get_query(info).filter( - SectorsModel.id == sector_id - ) - if not len(query.all()): - raise GraphQLError("Error, Invalid ID") - return query.all() + """Return a sector by its row ID""" + sector_id = kwargs.get('id', 1) + with app.app_context(): + query = Sectors.get_query(info).filter( + SectorsModel.id == sector_id + ) + if not len(query.all()): + raise GraphQLError("Error, Invalid ID") + return query.all() def resolve_get_sectors_by_sector(self, info, **kwargs): - """Return a list of sectors by its sector""" - sector = kwargs.get('sector', 'EMPTY') - query = Sectors.get_query(info).filter( - SectorsModel.sector == sector - ) - - if not len(query.all()): - raise GraphQLError("Error, Sector does not exist") - return query.all() + """Return a list of sectors by its sector""" + sector = kwargs.get('sector', 'EMPTY') + with app.app_context(): + query = Sectors.get_query(info).filter( + SectorsModel.sector == sector + ) + if not len(query.all()): + raise GraphQLError("Error, Sector does not exist") + return query.all() def resolve_get_sector_by_zone(self, info, **kwargs): - """Return a list of sectors by their zone""" - zone = kwargs.get('zone') - query = Sectors.get_query(info).filter( - SectorsModel.zone == zone - ) - if not len(query.all()): - raise GraphQLError("Error, Zone does not exist") - return query.all() + """Return a list of sectors by their zone""" + zone = kwargs.get('zone') + with app.app_context(): + query = Sectors.get_query(info).filter( + SectorsModel.zone == zone + ) + if not len(query.all()): + raise GraphQLError("Error, Zone does not exist") + return query.all() diff --git a/api/tests/test_auth_wrapper.py b/api/tests/test_auth_wrapper.py index 8f7df1653e..f32d781aaf 100644 --- a/api/tests/test_auth_wrapper.py +++ b/api/tests/test_auth_wrapper.py @@ -20,13 +20,7 @@ from app import app from queries import schema from models import Users, User_affiliations, Organizations -from functions.error_messages import error_not_an_admin -from functions.auth_functions import ( - is_super_admin, - is_admin, - is_user_write, - is_user_read -) +from backend.security_check import SecurityAnalysisBackend remove_seed() @@ -113,6 +107,7 @@ def user_role_test_db_init(): class TestUserRole(TestCase): def test_user_read_claim(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -121,7 +116,7 @@ def test_user_read_claim(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -137,13 +132,14 @@ def test_user_read_claim(self): { testUserClaims(org: ORG1, role: USER_READ) } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['testUserClaims'] assert executed['data']['testUserClaims'] == 'User Passed User Read Claim' def test_user_write_claim(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -152,7 +148,7 @@ def test_user_write_claim(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -168,13 +164,14 @@ def test_user_write_claim(self): { testUserClaims(org: ORG1, role: USER_WRITE) } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['testUserClaims'] assert executed['data']['testUserClaims'] == 'User Passed User Write Claim' def test_admin_claim(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -183,7 +180,7 @@ def test_admin_claim(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -199,13 +196,14 @@ def test_admin_claim(self): { testUserClaims(org: ORG1, role: ADMIN) } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['testUserClaims'] assert executed['data']['testUserClaims'] == 'User Passed Admin Claim' def test_super_admin_claim(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -214,7 +212,7 @@ def test_super_admin_claim(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -230,13 +228,14 @@ def test_super_admin_claim(self): { testUserClaims(org: ORG1, role: SUPER_ADMIN) } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['testUserClaims'] assert executed['data']['testUserClaims'] == 'User Passed Super Admin Claim' def test_user_not_admin(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -245,7 +244,7 @@ def test_user_not_admin(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -261,7 +260,7 @@ def test_user_not_admin(self): { testUserClaims(org: ORG1, role: ADMIN) } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['errors'] assert executed['errors'][0] assert executed['errors'][0]['message'] == 'Error, user is not an admin for that org' diff --git a/api/tests/test_cost_check.py b/api/tests/test_cost_check.py new file mode 100644 index 0000000000..93ef8eb946 --- /dev/null +++ b/api/tests/test_cost_check.py @@ -0,0 +1,139 @@ +import sys +import os +from os.path import dirname, join, expanduser, normpath, realpath + +import pytest +from graphene.test import Client + +from unittest import TestCase + +# This is the only way I could get imports to work for unit testing. +PACKAGE_PARENT = '..' +SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__)))) +sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT))) + +from manage import seed, remove_seed +seed() +from db import db +from app import app +from queries import schema +from models import Sectors, Groups +from backend.security_check import SecurityAnalysisBackend +remove_seed() + + +@pytest.fixture(scope='class') +def user_schema_test_db_init(): + db.init_app(app) + + with app.app_context(): + test_sector = Sectors( + id=1, + zone='ZO1', + sector='SO1' + ) + db.session.add(test_sector) + test_group = Groups( + id=1, + s_group='GO1', + sector_id=1 + ) + db.session.add(test_group) + db.session.commit() + + yield + + with app.app_context(): + Groups.query.delete() + Sectors.query.delete() + db.session.commit() + + +## +# This class of tests works within the 'createUser' api endpoint +@pytest.mark.usefixtures('user_schema_test_db_init') +class TestCostCheck(TestCase): + def test_valid_cost_query(self): + backend = SecurityAnalysisBackend() + client = Client(schema) + query = client.execute( + ''' + { + getSectorById(id: 1) { + groups { + edges { + node { + sectorId + } + } + } + } + } + ''', backend=backend) + result_refr = { + "data": { + "getSectorById": [ + { + "groups": { + "edges": [ + { + "node": { + "sectorId": 1 + } + } + ] + } + } + ] + } + } + self.assertDictEqual(result_refr, query) + + def test_invalid_cost_query(self): + backend = SecurityAnalysisBackend(10, 5) + client = Client(schema) + executed = client.execute( + ''' + { + getSectorById(id: 1) { + groups { + edges { + node { + sectorId + } + } + } + groups { + edges { + node { + sectorId + } + } + } + groups { + edges { + node { + sectorId + } + } + } + groups { + edges { + node { + sectorId + } + } + } + groups { + edges { + node { + sectorId + } + } + } + } + } + ''', backend=backend) + assert executed['errors'] + assert executed['errors'][0] + assert executed['errors'][0]['message'] == 'Query cost is too high' diff --git a/api/tests/test_depth_check.py b/api/tests/test_depth_check.py new file mode 100644 index 0000000000..6d148a01b1 --- /dev/null +++ b/api/tests/test_depth_check.py @@ -0,0 +1,143 @@ +import sys +import os +from os.path import dirname, join, expanduser, normpath, realpath + +import pytest +from graphene.test import Client + +from unittest import TestCase + +# This is the only way I could get imports to work for unit testing. +PACKAGE_PARENT = '..' +SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__)))) +sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT))) + +from manage import seed, remove_seed +seed() +from db import * +from app import app +from queries import schema +from models import Sectors, Groups +from backend.security_check import SecurityAnalysisBackend +remove_seed() + + +@pytest.fixture(scope='class') +def user_schema_test_db_init(): + db.init_app(app) + + with app.app_context(): + test_sector = Sectors( + id=1, + zone='ZO1', + sector='SO1' + ) + db.session.add(test_sector) + test_group = Groups( + id=1, + s_group='GO1', + sector_id=1 + ) + db.session.add(test_group) + db.session.commit() + + yield + + with app.app_context(): + Groups.query.delete() + Sectors.query.delete() + db.session.commit() + + +## +# This class of tests works within the 'createUser' api endpoint +@pytest.mark.usefixtures('user_schema_test_db_init') +class TestDepthCheck(TestCase): + def test_valid_depth_query(self): + backend = SecurityAnalysisBackend(10) + client = Client(schema) + query = client.execute( + ''' + { + getSectorById(id: 1) { + groups { + edges { + node { + sectorId + } + } + } + } + } + ''', backend=backend) + result_refr = { + "data": { + "getSectorById": [ + { + "groups": { + "edges": [ + { + "node": { + "sectorId": 1 + } + } + ] + } + } + ] + } + } + self.assertDictEqual(result_refr, query) + + def test_invalid_depth_query(self): + backend = SecurityAnalysisBackend(10) + client = Client(schema) + executed = client.execute( + ''' + { + getSectorById(id: 1) { + groups{ + edges{ + node{ + groupSector{ + groups{ + edges{ + node{ + groupSector{ + groups{ + edges{ + node{ + groupSector{ + groups{ + edges{ + node{ + groupSector{ + groups{ + edges{ + node{ + groupSector + } + } + } + } + } + } + } + } + } + } + } + } + } + } + } + } + } + } + } + } + } + ''', backend=backend) + assert executed['errors'] + assert executed['errors'][0] + assert executed['errors'][0]['message'] == 'Query is too complex' diff --git a/api/tests/test_domains_resolver.py b/api/tests/test_domains_resolver.py index 289949a1e8..82b683f49f 100644 --- a/api/tests/test_domains_resolver.py +++ b/api/tests/test_domains_resolver.py @@ -14,6 +14,7 @@ from db import db from models import Organizations, Domains from queries import schema +from backend.security_check import SecurityAnalysisBackend remove_seed() # This is the only way I could get imports to work for unit testing. @@ -55,6 +56,7 @@ class TestDomainsResolver(TestCase): def test_get_domain_resolvers_by_id(self): """Test get_domain_by_id resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -72,12 +74,13 @@ def test_get_domain_resolvers_by_id(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_domain_resolvers_by_domain(self): """"Test get_domain_by_domain resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -95,12 +98,13 @@ def test_get_domain_resolvers_by_domain(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_domain_resolvers_by_org(self): """Test get_domain_by_org_enum resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -118,12 +122,13 @@ def test_get_domain_resolvers_by_org(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_domain_resolver_by_id_invalid(self): """Test get_domain_by_id invalid ID error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -131,7 +136,7 @@ def test_domain_resolver_by_id_invalid(self): domain } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -140,6 +145,7 @@ def test_domain_resolver_by_id_invalid(self): def test_domain_resolver_by_url_invalid(self): """Test get_domain_by_domain invalid sector error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -147,7 +153,7 @@ def test_domain_resolver_by_url_invalid(self): domain } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -157,6 +163,7 @@ def test_domain_resolver_by_url_invalid(self): def test_domain_resolver_by_org_invalid(self): """Test get_domain_by_org invalid Zone error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -164,7 +171,7 @@ def test_domain_resolver_by_org_invalid(self): domain } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] diff --git a/api/tests/test_groups_resolver.py b/api/tests/test_groups_resolver.py index 48153e4bbf..21bc8f1920 100644 --- a/api/tests/test_groups_resolver.py +++ b/api/tests/test_groups_resolver.py @@ -14,6 +14,7 @@ from db import db from models import Sectors, Groups from queries import schema +from backend.security_check import SecurityAnalysisBackend remove_seed() # This is the only way I could get imports to work for unit testing. @@ -75,6 +76,7 @@ class TestGroupResolver(TestCase): def test_get_group_resolvers_by_id(self): """Test get_group_by_id resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -94,12 +96,13 @@ def test_get_group_resolvers_by_id(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_group_resolvers_by_group(self): """"Test get_group_by_group resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -119,12 +122,13 @@ def test_get_group_resolvers_by_group(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_group_resolvers_by_sector(self): """Test get_group_by_sector_id resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -150,12 +154,13 @@ def test_get_group_resolvers_by_sector(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_group_resolver_by_id_invalid(self): """Test get_group_by_id invalid ID error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -165,7 +170,7 @@ def test_group_resolver_by_id_invalid(self): sectorId } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -174,6 +179,7 @@ def test_group_resolver_by_id_invalid(self): def test_group_resolver_by_group_invalid(self): """Test get_group_by_group invalid sector error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -183,7 +189,7 @@ def test_group_resolver_by_group_invalid(self): sectorId } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -193,6 +199,7 @@ def test_group_resolver_by_group_invalid(self): def test_group_resolver_by_sector_invalid(self): """Test get_group_by_sector invalid Zone error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -202,7 +209,7 @@ def test_group_resolver_by_sector_invalid(self): sectorId } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] diff --git a/api/tests/test_notification_emails.py b/api/tests/test_notification_emails.py index dd037d5334..a0472f672c 100644 --- a/api/tests/test_notification_emails.py +++ b/api/tests/test_notification_emails.py @@ -13,7 +13,7 @@ seed() from queries import schema - +from backend.security_check import SecurityAnalysisBackend remove_seed() # This is the only way I could get imports to work for unit testing. @@ -27,6 +27,7 @@ def test_email_sent_successfully(self): """Test for ensuring that an email is sent to a valid email address""" request_headers = {'Origin': "https://testserver.com"} with app.test_request_context(headers=request_headers): + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -40,7 +41,7 @@ def test_email_sent_successfully(self): } } } - ''') + ''', backend=backend) assert executed['data'] assert executed['data']['sendPasswordReset'] @@ -64,6 +65,7 @@ def test_email_sent_successfully(self): def test_invalid_email(self): """Tests to ensure that an invalid email address will raise an error""" + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -72,7 +74,7 @@ def test_invalid_email(self): id } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] assert executed['errors'][0]['message'] == scalar_error_type( @@ -84,6 +86,7 @@ def test_email_sent_successfully(self): """Test for ensuring that an email is sent to a valid email address""" request_headers = {'Origin': "https://testserver.com"} with app.test_request_context(headers=request_headers): + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -97,7 +100,7 @@ def test_email_sent_successfully(self): } } } - ''') + ''', backend=backend) assert executed['data'] assert executed['data']['sendValidationEmail'] assert executed['data']['sendValidationEmail']['content'] @@ -121,6 +124,7 @@ def test_email_sent_successfully(self): def test_invalid_email(self): """Tests to ensure that an invalid email address will raise an error""" + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -129,7 +133,7 @@ def test_invalid_email(self): id } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] assert executed['errors'][0]['message'] == scalar_error_type( diff --git a/api/tests/test_organizations_resolver.py b/api/tests/test_organizations_resolver.py index 5d72bb66dd..ab2408b850 100644 --- a/api/tests/test_organizations_resolver.py +++ b/api/tests/test_organizations_resolver.py @@ -15,6 +15,7 @@ from db import db from models import Groups, Organizations from queries import schema +from backend.security_check import SecurityAnalysisBackend remove_seed() # This is the only way I could get imports to work for unit testing. @@ -57,6 +58,7 @@ class TestOrgResolver(TestCase): def test_get_org_resolvers_by_id(self): """Test get_organization_by_id resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -76,12 +78,13 @@ def test_get_org_resolvers_by_id(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_org_resolvers_by_org(self): """"Test get_org_by_org resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -101,12 +104,13 @@ def test_get_org_resolvers_by_org(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_org_resolvers_by_group(self): """Test get_org_by_group_id resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -126,12 +130,13 @@ def test_get_org_resolvers_by_group(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_org_resolver_by_id_invalid(self): """Test get_org_by_id invalid ID error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -140,7 +145,7 @@ def test_org_resolver_by_id_invalid(self): groupId } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -149,6 +154,7 @@ def test_org_resolver_by_id_invalid(self): def test_org_resolver_by_org_invalid(self): """Test get_org_by_org invalid sector error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -157,7 +163,7 @@ def test_org_resolver_by_org_invalid(self): description } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -166,6 +172,7 @@ def test_org_resolver_by_org_invalid(self): def test_org_resolver_by_group_invalid(self): """Test get_org_by_group invalid Zone error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -174,7 +181,7 @@ def test_org_resolver_by_group_invalid(self): description } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] diff --git a/api/tests/test_scans_resolver.py b/api/tests/test_scans_resolver.py index ab525b8d70..1819ab2013 100644 --- a/api/tests/test_scans_resolver.py +++ b/api/tests/test_scans_resolver.py @@ -16,6 +16,7 @@ from db import db from models import Scans, Domains, Users from queries import schema +from backend.security_check import SecurityAnalysisBackend remove_seed() # This is the only way I could get imports to work for unit testing. @@ -81,6 +82,7 @@ class TestScansResolver(TestCase): def test_get_scan_resolver_by_id(self): """Test get_sector_by_id resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -106,13 +108,14 @@ def test_get_scan_resolver_by_id(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_scans_by_date(self): """Test get_scans_by_date resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -138,13 +141,14 @@ def test_get_scans_by_date(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_scan_resolver_get_scans_by_date_range(self): """Test get_scans_by_date_range resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -176,13 +180,14 @@ def test_scan_resolver_get_scans_by_date_range(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_scan_resolver_get_scans_by_domain(self): """Test get_scans_by_domain resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -214,13 +219,14 @@ def test_scan_resolver_get_scans_by_domain(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_scan_resolver_get_scans_by_user_id(self): """Test get_scans_by_user_id resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -252,13 +258,14 @@ def test_scan_resolver_get_scans_by_user_id(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_scan_resolver_by_id_invalid(self): """Test get_scan_by_id invalid ID error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -268,7 +275,7 @@ def test_scan_resolver_by_id_invalid(self): } } """ - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -277,6 +284,7 @@ def test_scan_resolver_by_id_invalid(self): def test_scan_resolver_by_date_invalid(self): """Test get_scan_by_date invalid date error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -285,7 +293,7 @@ def test_scan_resolver_by_date_invalid(self): } } """ - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -294,6 +302,7 @@ def test_scan_resolver_by_date_invalid(self): def test_scan_resolver_by_date_range_invalid(self): """Test get_scan_by_date_range invalid date range error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -302,7 +311,7 @@ def test_scan_resolver_by_date_range_invalid(self): } } """ - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -311,6 +320,7 @@ def test_scan_resolver_by_date_range_invalid(self): def test_scan_resolver_by_nonexsiting_domain_invalid(self): """Test get_scan_by_domain invalid domain error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -319,7 +329,7 @@ def test_scan_resolver_by_nonexsiting_domain_invalid(self): } } """ - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -328,6 +338,7 @@ def test_scan_resolver_by_nonexsiting_domain_invalid(self): def test_scan_resolver_by_domain_invalid(self): """Test get_scan_by_domain no scan assocaited with that domain error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -336,7 +347,7 @@ def test_scan_resolver_by_domain_invalid(self): } } """ - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -345,6 +356,7 @@ def test_scan_resolver_by_domain_invalid(self): def test_scan_resolver_by_user_id_invalid_no_scans(self): """Test get_scan_by_user_id cannot find associated scans""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -353,7 +365,7 @@ def test_scan_resolver_by_user_id_invalid_no_scans(self): } } """ - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -362,6 +374,7 @@ def test_scan_resolver_by_user_id_invalid_no_scans(self): def test_scan_resolver_by_user_invalid_id(self): """Test get_scan_by_user_id cannot find id""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -370,7 +383,7 @@ def test_scan_resolver_by_user_invalid_id(self): } } """ - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] diff --git a/api/tests/test_sector_resolver.py b/api/tests/test_sector_resolver.py index c399b8412b..8692f4d54f 100644 --- a/api/tests/test_sector_resolver.py +++ b/api/tests/test_sector_resolver.py @@ -14,6 +14,7 @@ from app import app from models import Sectors from queries import schema +from backend.security_check import SecurityAnalysisBackend remove_seed() # This is the only way I could get imports to work for unit testing. @@ -65,6 +66,7 @@ class TestSectorResolver(TestCase): def test_get_sector_resolver_by_id(self): """Test get_sector_by_id resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -86,13 +88,14 @@ def test_get_sector_resolver_by_id(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_sector_resolver_by_sector(self): """Test get_sector_by_sector resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -112,13 +115,14 @@ def test_get_sector_resolver_by_sector(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_get_sector_resolver_by_zone(self): """Test get_sector_by_zone resolver""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -138,13 +142,14 @@ def test_get_sector_resolver_by_zone(self): } } - result_eval = client.execute(query) + result_eval = client.execute(query, backend=backend) self.assertDictEqual(result_refr, result_eval) def test_sector_resolver_by_id_invalid(self): """Test get_sector_by_id invalid ID error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -155,7 +160,7 @@ def test_sector_resolver_by_id_invalid(self): description } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -164,6 +169,7 @@ def test_sector_resolver_by_id_invalid(self): def test_sector_resolver_by_sector_invalid(self): """Test get_sector_by_sector invalid sector error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -173,7 +179,7 @@ def test_sector_resolver_by_sector_invalid(self): description } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -182,6 +188,7 @@ def test_sector_resolver_by_sector_invalid(self): def test_sector_resolver_by_zone_invalid(self): """Test get_sector_by_zone invalid Zone error handling""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) query = """ { @@ -192,7 +199,7 @@ def test_sector_resolver_by_zone_invalid(self): description } }""" - executed = client.execute(query) + executed = client.execute(query, backend=backend) assert executed['errors'] assert executed['errors'][0] diff --git a/api/tests/test_user_roles.py b/api/tests/test_user_roles.py index 5c88a52414..b434d72f32 100644 --- a/api/tests/test_user_roles.py +++ b/api/tests/test_user_roles.py @@ -21,12 +21,7 @@ from queries import schema from models import Users, User_affiliations, Organizations from functions.error_messages import error_not_an_admin -from functions.auth_functions import ( - is_super_admin, - is_admin, - is_user_write, - is_user_read -) +from backend.security_check import SecurityAnalysisBackend remove_seed() @@ -85,6 +80,7 @@ def user_role_test_db_init(): class TestUserUpdateWriteRole(TestCase): def test_user_claim_update_to_user_write(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -93,7 +89,7 @@ def test_user_claim_update_to_user_write(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -111,7 +107,7 @@ def test_user_claim_update_to_user_write(self): status } } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['updateUserRole'] assert executed['data']['updateUserRole']['status'] == 'Update Successful' @@ -124,7 +120,7 @@ def test_user_claim_update_to_user_write(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -140,7 +136,7 @@ def test_user_claim_update_to_user_write(self): { testUserClaims(org: ORG1, role: USER_WRITE) } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['testUserClaims'] assert executed['data']['testUserClaims'] == 'User Passed User Write Claim' @@ -150,6 +146,7 @@ def test_user_claim_update_to_user_write(self): class TestUserUpdateAdminRole(TestCase): def test_user_claim_update_to_admin(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -158,7 +155,7 @@ def test_user_claim_update_to_admin(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -176,7 +173,7 @@ def test_user_claim_update_to_admin(self): status } } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['updateUserRole'] assert executed['data']['updateUserRole']['status'] == 'Update Successful' @@ -189,7 +186,7 @@ def test_user_claim_update_to_admin(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -205,7 +202,7 @@ def test_user_claim_update_to_admin(self): { testUserClaims(org: ORG1, role: ADMIN) } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['testUserClaims'] assert executed['data']['testUserClaims'] == 'User Passed Admin Claim' @@ -215,6 +212,7 @@ def test_user_claim_update_to_admin(self): class TestUserUpdateSuperAdminRole(TestCase): def test_user_claim_update_to_super_admin(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -223,7 +221,7 @@ def test_user_claim_update_to_super_admin(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -241,7 +239,7 @@ def test_user_claim_update_to_super_admin(self): status } } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['updateUserRole'] assert executed['data']['updateUserRole']['status'] == 'Update Successful' @@ -254,7 +252,7 @@ def test_user_claim_update_to_super_admin(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -270,7 +268,7 @@ def test_user_claim_update_to_super_admin(self): { testUserClaims(org: ORG1, role: SUPER_ADMIN) } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['data'] assert executed['data']['testUserClaims'] assert executed['data']['testUserClaims'] == 'User Passed Super Admin Claim' @@ -280,6 +278,7 @@ def test_user_claim_update_to_super_admin(self): class TestUserUpdateAdminRoleInvalid(TestCase): def test_user_claim_update_to_user_write(self): with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) get_token = client.execute( ''' @@ -288,7 +287,7 @@ def test_user_claim_update_to_user_write(self): authToken } } - ''') + ''', backend=backend) assert get_token['data']['signIn']['authToken'] is not None token = get_token['data']['signIn']['authToken'] assert token is not None @@ -306,7 +305,7 @@ def test_user_claim_update_to_user_write(self): status } } - ''', context_value=request_headers) + ''', context_value=request_headers, backend=backend) assert executed['errors'] assert executed['errors'][0] assert executed['errors'][0]['message'] == error_not_an_admin() diff --git a/api/tests/test_user_schema.py b/api/tests/test_user_schema.py index 1d923bc511..cbe9508cfa 100644 --- a/api/tests/test_user_schema.py +++ b/api/tests/test_user_schema.py @@ -19,6 +19,7 @@ from app import app from queries import schema from models import Users +from backend.security_check import SecurityAnalysisBackend from functions.error_messages import * remove_seed() @@ -59,6 +60,7 @@ class TestCreateUser: def test_successful_creation(self): """Test that ensures a user can be created successfully using the api endpoint""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -71,7 +73,7 @@ def test_successful_creation(self): } } } - ''') + ''', backend=backend) assert executed['data'] assert executed['data']['createUser'] assert executed['data']['createUser']['user'] @@ -81,6 +83,7 @@ def test_successful_creation(self): def test_email_address_in_use(self): """Test that ensures each user has a unique email address""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) executed_first = client.execute( ''' @@ -92,7 +95,7 @@ def test_email_address_in_use(self): } } } - ''') + ''', backend=backend) executed = client.execute( ''' mutation{ @@ -103,7 +106,7 @@ def test_email_address_in_use(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -111,6 +114,7 @@ def test_email_address_in_use(self): def test_password_too_short(self): """Test that ensure that a user's password meets the valid length requirements""" + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -121,7 +125,7 @@ def test_password_too_short(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -129,6 +133,7 @@ def test_password_too_short(self): def test_passwords_do_not_match(self): """Test to ensure that user password matches their password confirmation""" + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -140,7 +145,7 @@ def test_passwords_do_not_match(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -154,6 +159,7 @@ class TestUpdatePassword: def test_update_password_success(self): """Test to ensure that a user is returned when their password is updated successfully""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -166,7 +172,7 @@ def test_update_password_success(self): } } } - ''') + ''', backend=backend) assert executed['data'] assert executed['data']['updatePassword'] @@ -176,6 +182,7 @@ def test_update_password_success(self): def test_updated_passwords_do_not_match(self): """Test to ensure that user's new password matches their password confirmation""" + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -187,7 +194,7 @@ def test_updated_passwords_do_not_match(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -195,6 +202,7 @@ def test_updated_passwords_do_not_match(self): def test_updated_password_too_short(self): """Test that ensure that a user's password meets the valid length requirements""" + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -205,7 +213,7 @@ def test_updated_password_too_short(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -213,6 +221,7 @@ def test_updated_password_too_short(self): def test_updated_password_no_user_email(self): """Test that ensures an empty string submitted as email will not be accepted""" + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -223,7 +232,7 @@ def test_updated_password_no_user_email(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -237,6 +246,7 @@ class TestValidateTwoFactor: def test_successful_validation(self): """Test that ensures a validation is successful when all params are proper""" with app.app_context(): + backend = SecurityAnalysisBackend() totp = pyotp.TOTP('base32secret3232') otp_code = totp.now() # Generates a code that is valid for 30s. Plenty of time to execute the query @@ -250,7 +260,7 @@ def test_successful_validation(self): } } } - ''') + ''', backend=backend) assert executed['data'] assert executed['data']['authenticateTwoFactor'] assert executed['data']['authenticateTwoFactor']['user'] @@ -259,6 +269,7 @@ def test_successful_validation(self): def test_user_does_not_exist(self): """Test that an error is raised if the user specified does not exist""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -269,7 +280,7 @@ def test_user_does_not_exist(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] @@ -278,6 +289,7 @@ def test_user_does_not_exist(self): def test_invalid_otp_code(self): """Test that an error is raised if the user specified does not exist""" with app.app_context(): + backend=SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -288,7 +300,7 @@ def test_invalid_otp_code(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] diff --git a/api/tests/test_user_sign_in.py b/api/tests/test_user_sign_in.py index 80d1adb161..717bb2c27f 100644 --- a/api/tests/test_user_sign_in.py +++ b/api/tests/test_user_sign_in.py @@ -19,6 +19,7 @@ from app import app from queries import schema from models import Users +from backend.security_check import SecurityAnalysisBackend from functions.error_messages import( error_invalid_credentials ) @@ -54,6 +55,7 @@ class TestSignInUser: def test_successful_sign_in(self): """Test that ensures a user can be signed in""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -64,7 +66,7 @@ def test_successful_sign_in(self): } } } - ''') + ''', backend=backend) assert executed['data'] assert executed['data']['signIn'] assert executed['data']['signIn']['user'] @@ -73,6 +75,7 @@ def test_successful_sign_in(self): def test_invalid_credentials(self): """Test that ensures a user can be signed in""" with app.app_context(): + backend = SecurityAnalysisBackend() client = Client(schema) executed = client.execute( ''' @@ -83,7 +86,7 @@ def test_invalid_credentials(self): } } } - ''') + ''', backend=backend) assert executed['errors'] assert executed['errors'][0] assert executed['errors'][0]['message']