diff --git a/api/Pipfile.lock b/api/Pipfile.lock index 4aefe38289..6f72580525 100644 --- a/api/Pipfile.lock +++ b/api/Pipfile.lock @@ -18,9 +18,9 @@ "default": { "alembic": { "hashes": [ - "sha256:2df2519a5b002f881517693b95626905a39c5faf4b5a1f94de4f1441095d1d26" + "sha256:791a5686953c4b366d3228c5377196db2f534475bb38d26f70eb69668efd9028" ], - "version": "==1.4.0" + "version": "==1.4.1" }, "aniso8601": { "hashes": [ @@ -237,9 +237,10 @@ }, "mako": { "hashes": [ - "sha256:2984a6733e1d472796ceef37ad48c26f4a984bb18119bb2dbc37a44d8f6e75a4" + "sha256:3139c5d64aa5d175dbafb95027057128b5fbd05a40c53999f3905ceb53366d9d", + "sha256:8e8b53c71c7e59f3de716b6832c4e401d903af574f6962edbbbf6ecc2a5fe6c9" ], - "version": "==1.1.1" + "version": "==1.1.2" }, "markupsafe": { "hashes": [ @@ -302,10 +303,10 @@ }, "packaging": { "hashes": [ - "sha256:170748228214b70b672c581a3dd610ee51f733018650740e98c7df862a583f73", - "sha256:e665345f9eef0c621aa0bf2f8d78cf6d21904eef16a93f020240b704a57f1334" + "sha256:3c292b474fda1671ec57d46d739d072bfd495a4f51ad01a055121d81e952b7a3", + "sha256:82f77b9bee21c1bafbf35a84905d604d5d1223801d639cf3ed140bd651c08752" ], - "version": "==20.1" + "version": "==20.3" }, "pluggy": { "hashes": [ @@ -382,9 +383,10 @@ }, "pycparser": { "hashes": [ - "sha256:a988718abfad80b6b157acce7bf130a30876d27603738ac39f140993246b25b3" + "sha256:2d475327684562c3a96cc71adf7dc8c4f0565175cf86b6d7a404ff4c771f15f0", + "sha256:7582ad22678f0fcd81102833f60ef8d0e57288b6b5fb00323d101be910e35705" ], - "version": "==2.19" + "version": "==2.20" }, "pyjwt": { "hashes": [ @@ -498,10 +500,10 @@ }, "zipp": { "hashes": [ - "sha256:12248a63bbdf7548f89cb4c7cda4681e537031eda29c02ea29674bc6854460c2", - "sha256:7c0f8e91abc0dc07a5068f315c52cb30c66bfbc581e5b50704c8a2f6ebae794a" + "sha256:aa36550ff0c0b7ef7fa639055d797116ee891440eac1a56f378e2d3179e0320b", + "sha256:c599e4d75c98f6798c509911d08a22e6c021d074469042177c8c86fb92eefd96" ], - "version": "==3.0.0" + "version": "==3.1.0" } }, "develop": { @@ -639,10 +641,10 @@ }, "packaging": { "hashes": [ - "sha256:170748228214b70b672c581a3dd610ee51f733018650740e98c7df862a583f73", - "sha256:e665345f9eef0c621aa0bf2f8d78cf6d21904eef16a93f020240b704a57f1334" + "sha256:3c292b474fda1671ec57d46d739d072bfd495a4f51ad01a055121d81e952b7a3", + "sha256:82f77b9bee21c1bafbf35a84905d604d5d1223801d639cf3ed140bd651c08752" ], - "version": "==20.1" + "version": "==20.3" }, "pluggy": { "hashes": [ @@ -746,10 +748,10 @@ }, "zipp": { "hashes": [ - "sha256:12248a63bbdf7548f89cb4c7cda4681e537031eda29c02ea29674bc6854460c2", - "sha256:7c0f8e91abc0dc07a5068f315c52cb30c66bfbc581e5b50704c8a2f6ebae794a" + "sha256:aa36550ff0c0b7ef7fa639055d797116ee891440eac1a56f378e2d3179e0320b", + "sha256:c599e4d75c98f6798c509911d08a22e6c021d074469042177c8c86fb92eefd96" ], - "version": "==3.0.0" + "version": "==3.1.0" } } } diff --git a/api/app/__init__.py b/api/app/__init__.py index d93afbe37a..93db228787 100644 --- a/api/app/__init__.py +++ b/api/app/__init__.py @@ -1,5 +1,7 @@ from flask import Flask +from app.logger import logger + from db import ( db, DB_NAME, diff --git a/api/app/logger.py b/api/app/logger.py new file mode 100644 index 0000000000..1048ddcbf2 --- /dev/null +++ b/api/app/logger.py @@ -0,0 +1,66 @@ +import logging +import logging.config +from graphql import GraphQLError + + +class GraphQLErrorLogFilter(logging.Filter): + def filter(self, record): + exc_type, exc, _ = record.exc_info + graphql_error = isinstance(exc, GraphQLError) + if graphql_error: + return False + return True + + +class GraphQLLocatedLogFilter(logging.Filter): + def filter(self, record): + if 'graphql.error.located_error.GraphQLLocatedError:' in record.msg: + return False + return True + + +class GraphQLTracebackLogFilter(logging.Filter): + def filter(self, record): + if 'graphql.execution.utils:Traceback (most recent call last):' in record.msg: + return False + return True + + +logger_dict = { + 'version': 1, + 'disable_existing_loggers': False, + 'handlers': { + 'console': { + 'level': 'DEBUG', + 'class': 'logging.StreamHandler', + }, + }, + # Prevent graphql exception from displaying in console + 'filters': { + 'graphql_error_log_filter': { + '()': GraphQLErrorLogFilter, + }, + 'graphql_located_error_filter': { + '()': GraphQLLocatedLogFilter + }, + 'graphql_traceback_error_filter': { + '()': GraphQLTracebackLogFilter + } + }, + 'loggers': { + 'graphql.execution.executor': { + 'level': 'WARNING', + 'handlers': ['console'], + 'filters': [ + 'graphql_error_log_filter', + 'graphql_located_error_filter', + 'graphql_traceback_error_filter', + ], + }, + }, +} + +logging.config.dictConfig(logger_dict) +logger = logging.getLogger('custom') +# logging.error('Error message') + diff --git a/api/functions/auth_wrappers.py b/api/functions/auth_wrappers.py index 8e46c2b6c6..c7b1a1f054 100644 --- a/api/functions/auth_wrappers.py +++ b/api/functions/auth_wrappers.py @@ -4,7 +4,7 @@ from graphql import GraphQLError from functions.orm_to_dict import orm_to_dict -from app import app +from app import app, logger from models import User_affiliations, Organizations user_admin_perm = ['super_admin', 'admin'] @@ -20,12 +20,16 @@ def decode_auth_token(request): """ auth_header = request.headers.get('Authorization') try: - payload = jwt.decode(auth_header, os.getenv('SUPER_SECRET_SALT'), algorithms=['HS256']) + payload = jwt.decode( + auth_header, + os.getenv('SUPER_SECRET_SALT'), + algorithms=['HS256'] + ) return payload['roles'] except jwt.ExpiredSignatureError: - raise GraphQLError('Signature expired. Please login again') + raise GraphQLError('Signature expired, please login again') except jwt.InvalidTokenError: - raise GraphQLError('Invalid token. Please login again') + raise GraphQLError('Invalid token, please login again') def check_user_claims(user_claims): @@ -36,14 +40,17 @@ def check_user_claims(user_claims): :param user_claims: A list of dicts that contain the users claims :return: Returns a valid list of user claims """ - if user_claims: + user_roles = [] + if user_claims[0] == 'none': + raise GraphQLError('Error, please contact your administrator for ' + 'organization access.') + elif user_claims: user_id = user_claims[0]['user_id'] with app.app_context(): user_aff = User_affiliations.query.filter( User_affiliations.user_id == user_id).all() user_aff = orm_to_dict(user_aff) if user_aff: - user_roles = [] for select in user_aff: temp_dict = { 'user_id': select['user_id'], @@ -57,13 +64,12 @@ def check_user_claims(user_claims): + list( itertools.filterfalse(lambda x: x in user_roles, user_claims)) if user_claim_diff: - print(user_claim_diff) # User has a difference in their claims - raise GraphQLError("Error, Please sign in again.") + raise GraphQLError("Error, please sign in again.") else: return user_claims else: - raise GraphQLError("User has no claims") + raise GraphQLError("User has no claims, please sign in again") def require_token(method): @@ -74,4 +80,5 @@ def wrapper(self, *args, **kwargs): kwargs['user_roles'] = user_claims return method(self, *args, **kwargs) raise GraphQLError(auth_resp) + return wrapper diff --git a/api/functions/db_seeding/organizations.py b/api/functions/db_seeding/organizations.py index ba1e9d3eb0..728f85e474 100644 --- a/api/functions/db_seeding/organizations.py +++ b/api/functions/db_seeding/organizations.py @@ -22,6 +22,14 @@ def seed_org(db, app): db.session.add(org) db.session.commit() + org = Organizations( + id=3, + organization='ORG3', + description='Organization 3', + ) + with app.app_context(): + db.session.add(org) + db.session.commit() def remove_org(db, app): with app.app_context(): diff --git a/api/functions/get_domain.py b/api/functions/get_domain.py new file mode 100644 index 0000000000..c9fc75e598 --- /dev/null +++ b/api/functions/get_domain.py @@ -0,0 +1,14 @@ +from sqlalchemy.orm import load_only + +from db import db +from models import Scans, Domains + + +def get_domain(self, info): + domain_id = db.session.query(Scans).filter( + Scans.id == self.id + ).options(load_only('domain_id')).first() + domain = db.session.query(Domains).filter( + Domains.id == domain_id.domain_id + ).options(load_only('domain')).first() + return domain.domain diff --git a/api/functions/get_timestamp.py b/api/functions/get_timestamp.py new file mode 100644 index 0000000000..c2b92c314b --- /dev/null +++ b/api/functions/get_timestamp.py @@ -0,0 +1,10 @@ +from sqlalchemy.orm import load_only + +from db import db +from models import Scans, Domains + +def get_timestamp(self, info): + timestamp = db.session.query(Scans).filter( + Scans.id == self.id + ).options(load_only('scan_date')).first() + return timestamp.scan_date diff --git a/api/functions/sign_in_user.py b/api/functions/sign_in_user.py index be11017aac..c39e5ec1d6 100644 --- a/api/functions/sign_in_user.py +++ b/api/functions/sign_in_user.py @@ -55,7 +55,7 @@ def sign_in_user(user_name, password): } user_roles.append(temp_dict) else: - user_roles = 'none' + user_roles = ['none'] try: payload = { 'exp': datetime.datetime.utcnow() + datetime.timedelta(days=0, seconds=1800), diff --git a/api/migrations/versions/c98ceb3d9163_.py b/api/migrations/versions/c98ceb3d9163_.py new file mode 100644 index 0000000000..5c55a35822 --- /dev/null +++ b/api/migrations/versions/c98ceb3d9163_.py @@ -0,0 +1,40 @@ +"""empty message + +Revision ID: c98ceb3d9163 +Revises: 48853738b19d +Create Date: 2020-03-09 08:13:31.523128 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'c98ceb3d9163' +down_revision = '48853738b19d' +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column('domains', 'scan_ssl') + op.drop_column('domains', 'scan_dmarc_psl') + op.drop_column('domains', 'scan_dkim') + op.drop_column('domains', 'scan_dmarc') + op.drop_column('domains', 'scan_https') + op.drop_column('domains', 'scan_mx') + op.drop_column('domains', 'scan_spf') + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.add_column('domains', sa.Column('scan_spf', sa.BOOLEAN(), autoincrement=False, nullable=True)) + op.add_column('domains', sa.Column('scan_mx', sa.BOOLEAN(), autoincrement=False, nullable=True)) + op.add_column('domains', sa.Column('scan_https', sa.BOOLEAN(), autoincrement=False, nullable=True)) + op.add_column('domains', sa.Column('scan_dmarc', sa.BOOLEAN(), autoincrement=False, nullable=True)) + op.add_column('domains', sa.Column('scan_dkim', sa.BOOLEAN(), autoincrement=False, nullable=True)) + op.add_column('domains', sa.Column('scan_dmarc_psl', sa.BOOLEAN(), autoincrement=False, nullable=True)) + op.add_column('domains', sa.Column('scan_ssl', sa.BOOLEAN(), autoincrement=False, nullable=True)) + # ### end Alembic commands ### diff --git a/api/models.py b/api/models.py index 7801a37c5f..768f3f211c 100644 --- a/api/models.py +++ b/api/models.py @@ -13,13 +13,6 @@ class Domains(db.Model): id = Column(Integer, primary_key=True) domain = Column(String) last_run = Column(DateTime) - scan_spf = Column(Boolean) - scan_dmarc = Column(Boolean) - scan_dmarc_psl = Column(Boolean) - scan_mx = Column(Boolean) - scan_dkim = Column(Boolean) - scan_https = Column(Boolean) - scan_ssl = Column(Boolean) dmarc_phase = Column(Integer) organization_id = Column(Integer, ForeignKey('organizations.id')) organization = relationship("Organizations", back_populates="domains", cascade="all, delete") @@ -92,11 +85,6 @@ class Scans(db.Model): scan_date = Column(DateTime) initiated_by = Column(Integer, ForeignKey('users.id')) domain = relationship("Domains", back_populates="scans", cascade="all, delete") - dmarc = relationship("Dmarc_scans", back_populates="dmarc_flagged_scan", cascade="all, delete") - dkim = relationship("Dkim_scans", back_populates="dkim_flagged_scan", cascade="all, delete") - spf = relationship("Spf_scans", back_populates="spf_flagged_scan", cascade="all, delete") - https = relationship("Https_scans", back_populates="https_flagged_scan", cascade="all, delete") - ssl = relationship("Ssl_scans", back_populates="ssl_flagged_scan", cascade="all, delete") class Dmarc_scans(db.Model): @@ -104,7 +92,6 @@ class Dmarc_scans(db.Model): id = Column(Integer, ForeignKey('scans.id'), primary_key=True) dmarc_scan = Column(JSONB) - dmarc_flagged_scan = relationship("Scans", back_populates="dmarc", cascade="all, delete") class Dkim_scans(db.Model): @@ -112,7 +99,6 @@ class Dkim_scans(db.Model): id = Column(Integer, ForeignKey('scans.id'), primary_key=True) dkim_scan = Column(JSONB) - dkim_flagged_scan = relationship("Scans", back_populates="dkim", cascade="all, delete") class Spf_scans(db.Model): @@ -120,7 +106,6 @@ class Spf_scans(db.Model): id = Column(Integer, ForeignKey('scans.id'), primary_key=True) spf_scan = Column(JSONB) - spf_flagged_scan = relationship("Scans", back_populates="spf", cascade="all, delete") class Https_scans(db.Model): @@ -128,7 +113,6 @@ class Https_scans(db.Model): id = Column(Integer, ForeignKey('scans.id'), primary_key=True) https_scan = Column(JSONB) - https_flagged_scan = relationship("Scans", back_populates="https", cascade="all, delete") class Ssl_scans(db.Model): @@ -136,7 +120,6 @@ class Ssl_scans(db.Model): id = Column(Integer, ForeignKey('scans.id'), primary_key=True) ssl_scan = Column(JSONB) - ssl_flagged_scan = relationship("Scans", back_populates="ssl", cascade="all, delete") class Ciphers(db.Model): diff --git a/api/queries.py b/api/queries.py index 2478ae84e4..436071cf4e 100644 --- a/api/queries.py +++ b/api/queries.py @@ -1,17 +1,11 @@ import graphene from graphene import relay -from graphene_sqlalchemy import SQLAlchemyConnectionField +from graphene_sqlalchemy import SQLAlchemyConnectionField, SQLAlchemyObjectType +from app import app -from model_enums.groups import GroupEnums from model_enums.organiztions import OrganizationsEnum -from model_enums.sectors import SectorEnums, ZoneEnums -from resolvers.domains import ( - resolve_get_domain_by_id, - resolve_get_domain_by_domain, - resolve_get_domain_by_organization -) from model_enums.roles import RoleEnums from schemas.user import ( @@ -21,30 +15,12 @@ UpdateUserPassword, ValidateTwoFactor ) + from schemas.user_affiliations import ( UpdateUserRole ) -from scalars.url import URL -from scalars.email_address import EmailAddress - -from schemas.sectors import Sectors -from schemas.groups import Groups -from schemas.organizations import Organizations -from schemas.domains import Domains -from schemas.scans import Scans - -from resolvers.sectors import ( - resolve_get_sector_by_id, - resolve_get_sectors_by_sector, - resolve_get_sector_by_zone -) -from resolvers.groups import ( - resolve_get_group_by_id, - resolve_get_group_by_group, - resolve_get_group_by_sector -) from resolvers.notification_emails import ( resolve_send_password_reset, resolve_send_validation_email @@ -57,154 +33,53 @@ from resolvers.user_affiliations import ( resolve_test_user_claims ) -from resolvers.organizations import ( - resolve_get_org_by_id, - resolve_get_org_by_org, - resolve_get_orgs_by_group -) -from resolvers.scans import ( - resolve_get_scan_by_id, - resolve_get_scans_by_date, - resolve_get_scans_by_date_range, - resolve_get_scans_by_domain, - resolve_get_scans_by_user_id -) -from resolvers.sectors import ( - resolve_get_sector_by_id, - resolve_get_sectors_by_sector, - resolve_get_sector_by_zone -) from resolvers.users import ( resolve_generate_otp_url, ) from scalars.email_address import EmailAddress from scalars.url import URL -from schemas.domains import Domains -from schemas.groups import Groups from schemas.notification_email import (NotificationEmail) -from schemas.organizations import Organizations -from schemas.scans import Scans -from schemas.sectors import Sectors + from schemas.user import ( - UserConnection, + UserObject, CreateUser, SignInUser, UpdateUserPassword, ValidateTwoFactor, ) +from schemas.domain import Domain +from resolvers.domains import ( + resolve_domain, + resolve_domains +) class Query(graphene.ObjectType): """The central gathering point for all of the GraphQL queries.""" node = relay.Node.Field() - all_users = SQLAlchemyConnectionField(UserConnection, sort=None) - # all_users = graphene.List(UserObject, failedAttempts=graphene.Int(), resolver=resolve_all_users) - # all_sectors = SQLAlchemyConnectionField(SectorsConnection, sort=None) - # all_groups = SQLAlchemyConnectionField(GroupsConnection, sort=None) - get_sector_by_id = graphene.List( - of_type=Sectors, - id=graphene.Argument(graphene.Int, required=True), - resolver=resolve_get_sector_by_id, - description="Allows selection of a sector from a given sector ID" - ) - get_sectors_by_sector = graphene.List( - of_type=Sectors, - sector=graphene.Argument(SectorEnums, required=True), - resolver=resolve_get_sectors_by_sector, - description="Allows selection of sector information from a given sector enum" - ) - get_sector_by_zone = graphene.List( - of_type=Sectors, - zone=graphene.Argument(ZoneEnums, required=True), - resolver=resolve_get_sector_by_zone, - description="Allows selection of all sectors from a given zone enum" - ) - get_group_by_id = graphene.List( - of_type=Groups, - id=graphene.Argument(graphene.Int, required=True), - resolver=resolve_get_group_by_id, - description="Allows selection of a group from a given group ID" - ) - get_group_by_group = graphene.List( - of_type=Groups, - group=graphene.Argument(GroupEnums, required=True), - resolver=resolve_get_group_by_group, - description="Allows the selection of group information from a given group enum" - ) - get_group_by_sector = graphene.List( - of_type=Groups, - sector=graphene.Argument(SectorEnums, required=True), - resolver=resolve_get_group_by_sector, - description="Allows selection of groups from a given sector enum" - ) - get_org_by_id = graphene.List( - of_type=Organizations, - id=graphene.Argument(graphene.Int, required=True), - resolver=resolve_get_org_by_id, - description="Allows the selection of an organization from a given ID" - ) - get_org_by_org = graphene.List( - of_type=Organizations, - org=graphene.Argument(OrganizationsEnum, required=True), - resolver=resolve_get_org_by_org, - description="Allows the selection of an organization from its given organization code" - ) - get_org_by_group = graphene.List( - of_type=Organizations, - group=graphene.Argument(GroupEnums, required=True), - resolver=resolve_get_orgs_by_group, - description="Allows the selection of organizations from a given group" - ) - get_domain_by_id = graphene.List( - of_type=Domains, - id=graphene.Argument(graphene.Int, required=True), - resolver=resolve_get_domain_by_id, - description="Allows the selection of a domain from a given ID" - ) - get_domain_by_domain = graphene.List( - of_type=Domains, - url=graphene.Argument(URL, required=True), - resolver=resolve_get_domain_by_domain, - description="Allows the selection of a domain from a given domain" - ) - get_domain_by_organization = graphene.List( - of_type=Domains, - org=graphene.Argument(OrganizationsEnum, required=True), - resolver=resolve_get_domain_by_organization, - description="Allows the selection of domains under an organization" - ) - get_scan_by_id = graphene.List( - of_type=Scans, - id=graphene.Argument(graphene.Int, required=True), - resolver=resolve_get_scan_by_id, - description="Allows the selection of a scan from a given scan ID" - ) - get_scans_by_date = graphene.List( - of_type=Scans, - date=graphene.Argument(graphene.Date, required=True), - resolver=resolve_get_scans_by_date, - description="Allows selection of scans on a given date" - ) - get_scans_by_date_range = graphene.List( - of_type=Scans, - startDate=graphene.Argument(graphene.Date, required=True), - endDate=graphene.Argument(graphene.Date, required=True), - resolver=resolve_get_scans_by_date_range, - description="Allows selection of scans from a given date range" - ) - get_scans_by_domain = graphene.List( - of_type=Scans, + users = SQLAlchemyConnectionField(UserObject._meta.connection, sort=None) + + domain = SQLAlchemyConnectionField( + Domain._meta.connection, url=graphene.Argument(URL, required=True), - resolver=resolve_get_scans_by_domain, - description="Allows selection of scans from a given URL" - ) - get_scans_by_user_id = graphene.List( - of_type=Scans, - id=graphene.Argument(graphene.Int, required=True), - resolver=resolve_get_scans_by_user_id, - description="Allows selection of scans initiated by a given user" - ) + sort=None, + description="Select information on a specific domain." + ) + with app.app_context(): + def resolve_domain(self, info, **kwargs): + return resolve_domain(self, info, **kwargs) + + domains = SQLAlchemyConnectionField( + Domain._meta.connection, + organization=graphene.Argument(OrganizationsEnum, required=False), + sort=None, + description="Select information on an organizations domains, or all " + "domains a user has access to. " + ) + with app.app_context(): + def resolve_domains(self, info, **kwargs): + return resolve_domains(self, info, **kwargs) generate_otp_url = graphene.String( email=graphene.Argument(EmailAddress, required=True), diff --git a/api/resolvers/README.md b/api/resolvers/README.md new file mode 100644 index 0000000000..bd1b6cf4e0 --- /dev/null +++ b/api/resolvers/README.md @@ -0,0 +1,270 @@ +## Resolvers +This directory contains various resolvers for resolving user queries, allowing the +enforcement of access control, and filtration. You can view the +[GraphQL Docs](https://graphql.org/learn/execution/) for a more in depth explanation. + +#### Example Resolver +```python +# File: api/resolvers/domains.py + +@require_token +def resolve_domains(self, info, **kwargs): + # Get Information passed in from kwargs + organization = kwargs.get('organization') + user_role = kwargs.get('user_roles') + + # Generate list of org's the user has access to + org_id_list = [] + for role in user_role: + org_id_list.append(role["org_id"]) + + if not org_id_list: + raise GraphQLError("Error, you have not been assigned to any organization") + + # Retrieve information based on query + query = Domain.get_query(info) + + if organization: + # Retrieve org id from organization enum + with app.app_context(): + org_orm = db.session.query(Organizations).filter( + Organizations.organization == organization + ).options(load_only('id')) + + # Check if org exists + if not len(org_orm.all()): + raise GraphQLError("Error, no organization associated with that enum") + + # Convert to int id + org_id = org_orm.first().id + + # Check if user is super admin, and if true return all domains belonging to + # that domain + if is_super_admin(user_role=user_role): + query_rtn = query.filter( + Domains.organization_id == org_id + ).all() + + # If org has no domains related to it + if not len(query_rtn): + raise GraphQLError( + "Error, no domains associated with that organization") + # If user fails super admin test + else: + # Check if user has permission to view org + if is_user_read(user_role, org_id): + query_rtn = query.filter( + Domains.organization_id == org_id + ).all() + else: + raise GraphQLError( + "Error, you do not have permission to view that organization") + + return query_rtn + else: + if is_super_admin(user_role=user_role): + query_rtn = query.all() + if not query_rtn: + raise GraphQLError("Error, no domains to view") + return query_rtn + else: + query_rtr = [] + for org_id in org_id_list: + if is_user_read(user_role, org_id): + tmp_query = query.filter( + Domains.organization_id == org_id + ).first() + query_rtr.append(tmp_query) + return query_rtr +``` +This example is a copy of our `domain(url: URL!)` resolver. This resolver requires that an argument be +included in the query setup but that will be covered later on in this document. Working from top to +bottom this file will take in the request sent in by the user, after its been cleared by the backend +and we can process the request. +

+##### Gather Information From Arguments +We first gather the arguments that have been passed in through `kwargs`. `user_roles` is a special +case because it has actually been passed to the resolver through the `@required_token` wrapper +function. +```python + # Get Information passed in from kwargs + organization = kwargs.get('organization') + user_role = kwargs.get('user_roles') +``` + +##### Generating User Access List +The next step of the resolver is to gather all of the org id's that the user has access to so we are +able to pass it into the auth check functions. This is simply done by iterating through the list of +dictionary's inside the `user_role` list. +```python + # Generate list of org's the user has access to + org_id_list = [] + for role in user_role: + org_id_list.append(role["org_id"]) +``` + +##### Gathering Initial Request +To be able to filter the results for each user, we first need to gather all the information that has +been requested in the query that was sent in by the user. We do this by requesting the information +through a `get_query()` statement, and passing in the `info` argument that was passed in, in the +query object. +```python + # Get initial Domain Query Object + query = Domain.get_query(info) +``` + +##### Check If Argument Has Been Filled +Because the `organiztion` argument is not a required field, we need to check to see if it was +included because it will depend on how we filter the results that are returned to the user. +```python + if organization: + ... + ... +``` + +##### Verify Organization Still Exists +We need to verify that the organization is still existing incase an admin has come along and removed +it without the user refreshing his data. If the organization is found we then also grab the ID for +later. +```python + # Retrieve org id from organization enum + with app.app_context(): + org_orm = db.session.query(Organizations).filter( + Organizations.organization == organization + ).options(load_only('id')) + + # Check if org exists + if not len(org_orm.all()): + raise GraphQLError("Error, no organization associated with that enum") + + # Convert to int id + org_id = org_orm.first().id +``` + +##### Check If The User Is A Super User +In our application we have a role of `Super User` any user with this role has access to all the +information that is retrievable through the API. For this reason all we need to do is to check +if they are a `Super User`, if it is found to be that they are a super user we then filter the +results based on the optional argument that was passed in. +```python + # Check if user is super admin, and if true return all domains belonging to + # that domain + if is_super_admin(user_role=user_role): + query_rtn = query.filter( + Domains.organization_id == org_id + ).all() + + # If org has no domains related to it + if not len(query_rtn): + raise GraphQLError("Error, no domains associated with that organization") +``` + +##### Check User Has Read Access +To verify that a user actually has the ability to view the information related to that organization +we use the `is_user_read(user_role, org_id)` check. This will ensure that the user has the proper +rights to view this information. We then filter it to only return the information that is related to +the requested organization. +```python + # Check if user has permission to view org + if is_user_read(user_role, org_id): + query_rtn = query.filter( + Domains.organization_id == org_id + ).all() + else: + raise GraphQLError("Error, you do not have permission to view that organization") +``` + +##### Returning Information +We then simply return the filtered information back to the Query class which then gets parsed and +sent to the user who requested it. +```python + return query_rtn +``` + +##### No Argument Sent (Super Admin) +If the user left the `organization` argument empty we will assume that they are requesting all the +domains that they have access to. To do this we follow the same steps as before, firstly we see +if the user has `Super User` access which then we send all the domains that belong to all the +organizations. +```python + if is_super_admin(user_role=user_role): + query_rtn = query.all() + if not query_rtn: + raise GraphQLError("Error, no domains to view") + return query_rtn +``` + +##### No Argument Sent (User Read) +If the user left the `organization` argument empty we will assume that they are requesting all the +domains that they have access to. Because we are checking for a user who does not have `Super User` +we will have to filter out the results based on the claims that they sent in. To accomplish this +we double check that they have access in the requested organization, and then filter the results +based on that organization, and append it to a list of ORM's that get passed back to the parser. +```python + query_rtr = [] + for org_id in org_id_list: + if is_user_read(user_role, org_id): + tmp_query = query.filter( + Domains.organization_id == org_id + ).first() + query_rtr.append(tmp_query) + return query_rtr +``` + +### Query Class Design +```python +# File api/queries.py +class Query(graphene.ObjectType): + """The central gathering point for all of the GraphQL queries.""" + + domains = SQLAlchemyConnectionField( + Domain._meta.connection, + organization=graphene.Argument(OrganizationsEnum, required=False), + sort=None, + description="Select information on an organizations domains, or all " + "domains a user has access to. " + ) +``` +This is a small piece from our `Query` class that includes the important information for adding +a resolver for a query. To actually have the ability to query a relay connection we need to use +the `SQLAlchemyConnectionField` class and pass in some information. +#### Creating Query +##### Set Schema Model +First we need to send it the actual `SQLAlchemyObjectType` class that we are using, we use the +`._meta.connection` to ensure that we do not have a duplication error from the GraphQL server-core. +```python +Domain._meta.connection, +``` +##### Set Arguments +We then create an argument, this can be of any type that pertains to the request you would like to +create, here we are asking for an `organization` of type `OrganizationsEnm` this will limit the +user to send a request if there is actually an enum for that organization. To make this optional as +we have defined in our resolver function we set the `required=False` property to false. +```python +organization=graphene.Argument(OrganizationsEnum, required=False) +``` +##### Set Sorting +You are also able to sort this in any manner that you want, but for now we are just going to leave +it set to `None`. +```python +sort=None +``` + +##### Set Description +To have text generate for this object in the GraphQLi Docs we need to add the `description` property +and set a string value of what we want the description to say. +```python +description="Select information on an organizations domains, or all domains a user has access to." +``` + +#### Creating Resolver +```python + with app.app_context(): + def resolve_domains(self, info, **kwargs): + return resolve_domains(self, info, **kwargs) +``` +To resolve the query with our filtered material we add this statement in right after the connection +object that we just created. We use `with app.app_context()` to ensure that there are no database +session errors occur. We then name the function `resolve_(name of List, Field, or Connection)` +this will then automatically connect the two and use it to resolve. We use the `self, info, **kwargs` +as our arguments for the object so that they can be passed into the resolver function for our use. diff --git a/api/resolvers/domains.py b/api/resolvers/domains.py index cc086d23c0..abfbfe79bb 100644 --- a/api/resolvers/domains.py +++ b/api/resolvers/domains.py @@ -2,59 +2,148 @@ from sqlalchemy.orm import load_only from app import app +from db import db + +from functions.auth_wrappers import require_token +from functions.auth_functions import is_super_admin, is_user_read + +from models import Domains, Organizations + +from schemas.domain import Domain + + +@require_token +def resolve_domain(self: Domain, info, **kwargs): + """ + This function is to resolve the domain query which takes in a url as a URL + scalar and checks against the user roles passed in by the Authorization + header, and return filtered results depending on the users access level + :param self: Domain SQLAlchemyObject type defined in the schemas directory + :param info: Request information sent to the sever from a client + :param kwargs: Field arguments (i.e. url), and user_roles + :return: Filtered Domain SQLAlchemyObject Type + """ + # Get Information passed in via kwargs + url = kwargs.get('url') + user_role = kwargs.get('user_roles') + + # Generate list of org's the user has access to + org_id_list = [] + for role in user_role: + org_id_list.append(role["org_id"]) + + # Get initial Domain Query Object + query = Domain.get_query(info) + + # Check to see if the user is a super admin, if true return all information + if is_super_admin(user_role=user_role): + query_rtn = query.filter( + Domains.domain == url + ).all() + # If url cannot be found, raise error + if not len(query_rtn): + raise GraphQLError("Error, domain does not exist") + # If user fails super user test + else: + # Get org id that is related to the domain + org_orm = db.session.query(Organizations).filter( + Organizations.id == Domains.organization_id + ).filter( + Domains.domain == url + ).first() + + # If org cannot be found + if not org_orm: + raise GraphQLError("Error, domain does not exist") + org_id = org_orm.id + + # Check if user has read access or higher to the requested organization + if is_user_read(user_role, org_id): + query_rtn = query.filter( + Domains.domain == url + ).all() + else: + raise GraphQLError("Error, you do not have permission to view that domain") + + return query_rtn + + +@require_token +def resolve_domains(self, info, **kwargs): + """ + This function is to resolve the domain query which takes in a url as a URL + scalar and checks against the user roles passed in by the Authorization + header, and return filtered results depending on the users access level + :param self: Domain SQLAlchemyObject type defined in the schemas directory + :param info: Request information sent to the sever from a client + :param kwargs: Field arguments (i.e. organization), and user_roles + :return: Filtered Domain SQLAlchemyObject Type + """ + # Get Information passed in from kwargs + organization = kwargs.get('organization') + user_role = kwargs.get('user_roles') + + # Generate list of org's the user has access to + org_id_list = [] + for role in user_role: + org_id_list.append(role["org_id"]) + + if not org_id_list: + raise GraphQLError("Error, you have not been assigned to any organization") + + # Retrieve information based on query + query = Domain.get_query(info) + + if organization: + # Retrieve org id from organization enum + with app.app_context(): + org_orm = db.session.query(Organizations).filter( + Organizations.organization == organization + ).options(load_only('id')) + + # Check if org exists + if not len(org_orm.all()): + raise GraphQLError("Error, no organization associated with that enum") + + # Convert to int id + org_id = org_orm.first().id + + # Check if user is super admin, and if true return all domains belonging to + # that domain + if is_super_admin(user_role=user_role): + query_rtn = query.filter( + Domains.organization_id == org_id + ).all() + + # If org has no domains related to it + if not len(query_rtn): + raise GraphQLError( + "Error, no domains associated with that organization") + # If user fails super admin test + else: + # Check if user has permission to view org + if is_user_read(user_role, org_id): + query_rtn = query.filter( + Domains.organization_id == org_id + ).all() + else: + raise GraphQLError( + "Error, you do not have permission to view that organization") + + return query_rtn + else: + if is_super_admin(user_role=user_role): + query_rtn = query.all() + if not query_rtn: + raise GraphQLError("Error, no domains to view") + return query_rtn + else: + query_rtr = [] + for org_id in org_id_list: + if is_user_read(user_role, org_id): + tmp_query = query.filter( + Domains.organization_id == org_id + ).first() + query_rtr.append(tmp_query) + return query_rtr -from schemas.domains import ( - Domains, - DomainModel -) - -from schemas.organizations import ( - Organizations, - OrganizationsModel -) - - -# Resolvers -def resolve_get_domain_by_id(self, info, **kwargs): - """Return a domain by its row ID""" - group_id = kwargs.get('id', 1) - with app.app_context(): - query = Domains.get_query(info).filter( - DomainModel.id == group_id - ) - if not len(query.all()): - raise GraphQLError("Error, Invalid ID") - return query.all() - - -def resolve_get_domain_by_domain(self, info, **kwargs): - """Return a domain by a url""" - domain = kwargs.get('url') - with app.app_context(): - query = Domains.get_query(info).filter( - DomainModel.domain == domain - ) - if not len(query.all()): - raise GraphQLError("Error, domain does not exist") - return query.all() - - -def resolve_get_domain_by_organization(self, info, **kwargs): - """Return a list of domains by by their associated organization""" - organization = kwargs.get('org') - with app.app_context(): - organization_id = Organizations.get_query(info).filter( - OrganizationsModel.organization == organization - ).options(load_only('id')) - - if not len(organization_id.all()): - raise GraphQLError("Error, no organization associated with that enum") - with app.app_context(): - query = Domains.get_query(info).filter( - DomainModel.organization_id == organization_id - ) - - if not len(query.all()): - raise GraphQLError( - "Error, no domains associated with that organization") - return query.all() diff --git a/api/resolvers/groups.py b/api/resolvers/groups.py deleted file mode 100644 index 44409aede7..0000000000 --- a/api/resolvers/groups.py +++ /dev/null @@ -1,59 +0,0 @@ -from graphql import GraphQLError -from sqlalchemy.orm import load_only - -from app import app - -from schemas.groups import ( - Groups, - GroupsModel -) - -from schemas.sectors import ( - Sectors, - SectorsModel -) - - -# Resolvers -def resolve_get_group_by_id(self, info, **kwargs): - """Return a group by its row ID""" - group_id = kwargs.get('id', 1) - with app.app_context(): - query = Groups.get_query(info).filter( - GroupsModel.id == group_id - ) - if not len(query.all()): - raise GraphQLError("Error, Invalid ID") - return query.all() - - -def resolve_get_group_by_group(self, info, **kwargs): - """Return a list of groups by its group""" - group = kwargs.get('group') - with app.app_context(): - query = Groups.get_query(info).filter( - GroupsModel.s_group == group - ) - if not len(query.all()): - raise GraphQLError("Error, group does not exist ") - return query.all() - - -def resolve_get_group_by_sector(self, info, **kwargs): - """Return a list of groups by by their associated sector""" - sector = kwargs.get('sector') - with app.app_context(): - sector_id = Sectors.get_query(info).filter( - SectorsModel.sector == sector - ).options(load_only('id')) - - if not len(sector_id.all()): - raise GraphQLError("Error, no sector associated with that enum") - with app.app_context(): - query = Groups.get_query(info).filter( - GroupsModel.sector_id == sector_id - ) - - if not len(query.all()): - raise GraphQLError("Error, no groups associated with that sector") - return query.all() diff --git a/api/resolvers/organizations.py b/api/resolvers/organizations.py deleted file mode 100644 index 77f78edcfb..0000000000 --- a/api/resolvers/organizations.py +++ /dev/null @@ -1,58 +0,0 @@ -from graphql import GraphQLError -from sqlalchemy.orm import load_only - -from app import app - -from schemas.organizations import ( - Organizations, - OrganizationsModel -) - -from schemas.groups import ( - Groups, - GroupsModel -) - - -# Resolvers -def resolve_get_org_by_id(self, info, **kwargs): - """Return an organization by its id""" - org_id = kwargs.get('id') - with app.app_context(): - query = Organizations.get_query(info).filter( - OrganizationsModel.id == org_id - ) - if not len(query.all()): - raise GraphQLError("Error, Invalid ID") - return query.all() - - -def resolve_get_org_by_org(self, info, **kwargs): - """Return an organization by its organization code""" - org_code = kwargs.get('org') - with app.app_context(): - query = Organizations.get_query(info).filter( - OrganizationsModel.organization == org_code - ) - if not len(query.all()): - raise GraphQLError("Error, Invalid Organization") - return query.all() - - -def resolve_get_orgs_by_group(self, info, **kwargs): - group = kwargs.get('group') - with app.app_context(): - group_id = Groups.get_query(info).filter( - GroupsModel.s_group == group - ).options(load_only('id')) - - if not len(group_id.all()): - raise GraphQLError("Error, no group associated with that enum") - with app.app_context(): - query = Organizations.get_query(info).filter( - OrganizationsModel.group_id == group_id - ) - - if not len(query.all()): - raise GraphQLError("Error, no organizations associated with that group") - return query.all() diff --git a/api/resolvers/scans.py b/api/resolvers/scans.py deleted file mode 100644 index fc7a8c4f05..0000000000 --- a/api/resolvers/scans.py +++ /dev/null @@ -1,101 +0,0 @@ -from graphql import GraphQLError -from sqlalchemy.orm import load_only -import sqlalchemy as sq -from sqlalchemy import between - -from app import app - -from schemas.scans import ( - Scans, - ScanModel -) - -from schemas.domains import ( - Domains, - DomainModel -) - -from schemas.user import ( - User as UserModel, - UserObject as User -) - - -# Resolvers -def resolve_get_scan_by_id(self, info, **kwargs): - """Return a scan by its ID""" - scan_id = kwargs.get('id') - with app.app_context(): - query = Scans.get_query(info).filter( - ScanModel.id == scan_id - ) - if not len(query.all()): - raise GraphQLError("Error, Invalid ID") - return query.all() - - -def resolve_get_scans_by_date(self, info, **kwargs): - """Return a list of scans on a certain date""" - date = kwargs.get('date') - with app.app_context(): - query = Scans.get_query(info).filter( - sq.func.date_trunc('day', ScanModel.scan_date) == date - ) - if not len(query.all()): - raise GraphQLError("Error, No scans occurred on that date") - return query.all() - - -def resolve_get_scans_by_date_range(self, info, **kwargs): - """Return a list of scans from a date range""" - start_date = kwargs.get('startDate') - end_date = kwargs.get('endDate') - with app.app_context(): - query = Scans.get_query(info).filter( - sq.func.date_trunc('day', ScanModel.scan_date) >= start_date - ).filter( - sq.func.date_trunc('day', ScanModel.scan_date) <= end_date - ) - if not len(query.all()): - raise GraphQLError("Error, No scans in that date range") - return query.all() - - -def resolve_get_scans_by_domain(self, info, **kwargs): - """Return a list of scans based on a domain id""" - domain = kwargs.get('url') - with app.app_context(): - domain_id = Domains.get_query(info).filter( - DomainModel.domain == domain - ).options(load_only('id')) - - if not len(domain_id.all()): - raise GraphQLError("Error, no domain associated with that URL") - with app.app_context(): - query = Scans.get_query(info).filter( - ScanModel.domain_id == domain_id - ) - - if not len(query.all()): - raise GraphQLError("Error, no scans associated with that domain") - return query.all() - - -def resolve_get_scans_by_user_id(self, info, **kwargs): - """Return a list of scans based on a user id""" - user = kwargs.get('id') - with app.app_context(): - user_id = User.get_query(info).filter( - UserModel.id == user - ).options(load_only('id')) - - if not len(user_id.all()): - raise GraphQLError("Error, cannot find user") - with app.app_context(): - query = Scans.get_query(info).filter( - ScanModel.initiated_by == user_id - ) - - if not len(query.all()): - raise GraphQLError("Error, no scans initiated by that user") - return query.all() diff --git a/api/resolvers/sectors.py b/api/resolvers/sectors.py deleted file mode 100644 index 6303e10b99..0000000000 --- a/api/resolvers/sectors.py +++ /dev/null @@ -1,41 +0,0 @@ -from graphql import GraphQLError -from schemas.sectors import Sectors, SectorsModel -from model_enums.sectors import SectorEnums -from app 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) - 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') - 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') - 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/schemas/README.md b/api/schemas/README.md new file mode 100644 index 0000000000..484a359cc3 --- /dev/null +++ b/api/schemas/README.md @@ -0,0 +1,234 @@ +# Schemas +This directory contains individual files for each type that you may come across +in our [GraphQL](https://graphql.org/) API. + +### Writing A Single Object +```python +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType +from sqlalchemy.orm import load_only + +from app import app +from db import db +from models import ExampleModel + +class SingleObject(SQLAlchemyObjectType): + class Meta: + model = ExampleModel + exclude_fields = ( + "id", "field_1", + "field_2", + ) + id = graphene.ID() + field_1 = graphene.String() + new_field_1 = graphene.Int() + + with app.app_context(): + def resolve_id(self: ExampleModel, info): + return self.id + + def resolve_field_1(self: ExampleModel, info): + return self.field_1 + + def resolve_new_field_1(self: ExampleModel, info): + new_field = db.session.query(NewModel).filter( + NewModel.ExampleModel_Id == self.id + ).options(load_only("select_field).first() + return new_field.select_field +``` +This is an example to write a single type object. We use an `SQLAlchemyObjectType` because +it gives us the ability to include this type as a `graphene.List()` in another type +and be related to that entry. One interesting thing is that the self object that is being +passed into the resolve is actually an instance of the model that you are querying. + +--- + +### Writing A Single Object That Contains Another Single Object +```python +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType +from sqlalchemy.orm import load_only + +from app import app +from db import db +from models import ExampleModel, ExampleModel2 + + +class SingleObject(SQLAlchemyObjectType): + class Meta: + model = ExampleModel + exclude_fields = ( + "id", "field_1", + "field_2", + ) + id = graphene.ID() + field_1 = graphene.String() + new_field_1 = graphene.Int() + + with app.app_context(): + def resolve_id(self: ExampleModel, info): + return self.id + + def resolve_field_1(self: ExampleModel, info): + return self.field_1 + + def resolve_new_field_1(self: ExampleModel, info): + new_field = db.session.query(NewModel).filter( + NewModel.ExampleModel_Id == self.id + ).options(load_only("select_field).first() + return new_field.select_field + + +class ListObject(SQLAlchemyObjectType): + class Meta: + model = ExampleModel2 + exclude_fields = ( + "id", "field_1", + ) + id = graphene.ID() + new_field_1 = graphene.Int() + single_object = graphene.List(lambda: SingleObject) + + with app.app_context(): + def resolve_id(self: ExampleModel2, info): + return self.id + + def resolve_field_1(self: ExampleModel2, info): + return self.field_1 + + def resolve_new_field_1(self: ExampleModel2, info): + new_field = db.session.query(NewModel).filter( + NewModel.ExampleModel_Id == self.id + ).options(load_only("select_field).first() + return new_field.select_field + + def resolve_single_object(self: ExampleModel2, info): + query = SingleObject.get_query(info) + return query.all() +``` +If you are creating just a single object type, but want to include a single object type +as a sub-selection you need to create a new field of it as type `graphene.List()` you +then use a `lambda` and operate that on the object you want to have go in that place. +To resolve the fields you must have all the resovlers you want for that type completed +in its original definition. Then you simply pass the info to that object `query = SingleObject.get_query(info)` + +--- + +### Writing An Object With A Node +```python +import graphene +from graphene import relay +from graphene_sqlalchemy import SQLAlchemyObjectType +from sqlalchemy.orm import load_only + +from app import app +from db import db +from models import ExampleModel, ExampleModel2 + + +class NodeQueryObject(SQLAlchemyObjectType): + class Meta: + model = ExampleModel + interfaces = (relay.Node, ) + exclude_fields = ( + "id", "field_1", + "field_2", + ) + id = graphene.ID() + field_1 = graphene.String() + new_field_1 = graphene.Int() + + with app.app_context(): + def resolve_id(self: ExampleModel, info): + return self.id + + def resolve_field_1(self: ExampleModel, info): + return self.field_1 + + def resolve_new_field_1(self: ExampleModel, info): + new_field = db.session.query(NewModel).filter( + NewModel.ExampleModel_Id == self.id + ).options(load_only("select_field).first() + return new_field.select_field + + +class NodeQueryObjectConnection(relay.Connection): + class Meta: + node = NodeQueryObject + + +class ListObject(SQLAlchemyObjectType): + class Meta: + model = ExampleModel2 + exclude_fields( + "id", "field_1", + "field_2", + ) + id = graphene.ID() + field_1 = graphene.String() + new_field_1 = graphene.Int() + single_object = graphene.ConnectionField(NodeQueryObject._meta.connection) + + with app.app_context(): + def resolve_id(self: ExampleModel2, info): + return self.id + + def resolve_field_1(self: ExampleModel2, info): + return self.field_1 + + def resolve_new_field_1(self: ExampleModel2, info): + new_field = db.session.query(NewModel).filter( + NewModel.ExampleModel_Id == self.id + ).options(load_only("select_field).first() + return new_field.select_field + + def resolve_single_object(self: ExampleModel2, info): + query = SingleObject.get_query(info) + return query.all() +``` +##### Creating Node +To build a relay Node using `SQLAlchemyObjectType` we need to create a connection class that we can connect the +original object to. We need to add an interfaces field in the original objects +`class Meta:` -> `interfaces = (relay.Node, )`. Adding this will inform that the +object it is waiting for a connection class. To create the actual relay node we need +to create a `class ObjectNameConnection(relay.Connection)` object. In this connection +object we need to add a `class Meta:` again but this time with a `node = NodeQueryObject` +this will now connect the node with a `relay.Connection()`. + +##### Adding Node Into Object +To include a relay node in +a object type we use a `graphene.ConnectionField()` and just put in the object +`NodeQueryObject._meta.connction` we use the `._meta.connection`to help the schema understand +which connection belongs to which objects or else we may encounter a conflict where two +objects have the same connection name. + +--- + +### Adding Objects And Relays To Schema +```python +import graphene +from graphene_sqlalchemy import SQLAlchemyConnectionField + +from schemas.example import NodeQueryObject, ListObject + + +class Query(graphene.ObjectType): + node_query_object = SQLAlchemyConnectionField(NodeQueryObject._meta.connection) + list_object = graphene.List(lambda: ListObject) + + def resolve_list_object(self, info): + query = ListObject.get_query(info) + return query.all() + +class Mutation(graphene.ObjectType): + ... + ... + + +schema = graphene.Schema(query=Query, mutation=Mutation) +``` +To be able to access these new object types they have to be added to the schema. +If the object is being implemented as a `Node` we want to use the `SQLAlchemyConnectionField` +type with the `._meta.connection`. This will allow you to query this as a `relay.Node`. If you +want to a single object type, it follows the same method as if you were adding it to another +class via the `graphene.List(lambda: Object)` with a resolver querying the `Object.get_query(info)`. diff --git a/api/schemas/dkim.py b/api/schemas/dkim.py deleted file mode 100644 index ffefc90977..0000000000 --- a/api/schemas/dkim.py +++ /dev/null @@ -1,16 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType - -from models import Dkim_scans as DkimModel - - -class Dkim(SQLAlchemyObjectType): - class Meta: - model = DkimModel - interfaces = (relay.Node,) - - -class DkimConnection(relay.Connection): - class Meta: - node = Dkim diff --git a/api/schemas/dmarc.py b/api/schemas/dmarc.py deleted file mode 100644 index d8c44e340f..0000000000 --- a/api/schemas/dmarc.py +++ /dev/null @@ -1,16 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType - -from models import Dmarc_scans as DmarcModel - - -class Dmarc(SQLAlchemyObjectType): - class Meta: - model = DmarcModel - interfaces = (relay.Node,) - - -class DmarcConnection(relay.Connection): - class Meta: - node = Dmarc diff --git a/api/schemas/domain/__init__.py b/api/schemas/domain/__init__.py new file mode 100644 index 0000000000..f80d5dd35b --- /dev/null +++ b/api/schemas/domain/__init__.py @@ -0,0 +1,53 @@ +import graphene +from graphene import relay +from graphene_sqlalchemy import SQLAlchemyObjectType +from graphene_sqlalchemy.types import ORMField + +from app import app +from models import Domains +from scalars.url import URL + +from schemas.domain.email_scan import EmailScan +from schemas.domain.www_scan import WWWScan + + +class Domain(SQLAlchemyObjectType): + class Meta: + model = Domains + interfaces = (relay.Node, ) + exclude_fields = ( + "id", + "domain", + "last_run", + "dmarc_phase", + "organization_id", + "organization", + "scans" + ) + url = URL(description="The domain the scan was run on") + email = graphene.ConnectionField( + EmailScan._meta.connection, + description="DKIM, DMARC, and SPF scan results" + ) + www = graphene.ConnectionField( + WWWScan._meta.connection, + description="HTTPS, and SSL scan results" + ) + organization = ORMField(model_attr='organization') + + with app.app_context(): + def resolve_url(self: Domains, info): + return self.domain + + def resolve_email(self, info): + query = EmailScan.get_query(info) + return query.all() + + def resolve_www(self, info): + query = WWWScan.get_query(info) + return query.all() + + +class DomainConnection(relay.Connection): + class Meta: + node = Domain diff --git a/api/schemas/domain/email_scan/__init__.py b/api/schemas/domain/email_scan/__init__.py new file mode 100644 index 0000000000..6832c2f2ff --- /dev/null +++ b/api/schemas/domain/email_scan/__init__.py @@ -0,0 +1,69 @@ +import graphene +from graphene import relay +from graphene_sqlalchemy import SQLAlchemyObjectType +from graphql import ResolveInfo + +from app import app +from models import Scans, Dmarc_scans, Spf_scans, Dkim_scans +from scalars.url import URL + +from schemas.domain.email_scan.dkim import DKIM +from schemas.domain.email_scan.dmarc import DMARC +from schemas.domain.email_scan.spf import SPF +from functions.get_domain import get_domain +from functions.get_timestamp import get_timestamp + + +class EmailScan(SQLAlchemyObjectType): + """ + Results of DKIM, DMARC, and SPF scans, on domains + """ + class Meta: + model = Scans + interfaces = (relay.Node, ) + exclude_fields = ( + "id", + "domain_id", + "scan_date", + "initiated_by" + ) + domain = URL(description="The domain the scan was run on") + timestamp = graphene.DateTime(description="The time the scan was initiated") + dmarc = graphene.List( + lambda: DMARC, + description="Domain-based Message Authentication, Reporting, " + "and Conformance (DMARC) " + ) + spf = graphene.List( + lambda: SPF, + description="Sender Policy Framework (SPF) for Authorizing Use of " + "Domains in Email " + ) + dkim = graphene.List( + lambda: DKIM, + description="DomainKeys Identified Mail (DKIM) Signatures" + ) + + with app.app_context(): + def resolve_domain(self: Scans, info): + return get_domain(self, info) + + def resolve_timestamp(self: Scans, info): + return get_timestamp(self, info) + + def resolve_dmarc(self: Scans, info): + query = DMARC.get_query(info) + return query.filter(self.id == Dmarc_scans.id).all() + + def resolve_spf(self: Scans, info): + query = SPF.get_query(info) + return query.filter(self.id == Spf_scans.id).all() + + def resolve_dkim(self: Scans, info): + query = DKIM.get_query(info) + return query.filter(self.id == Dkim_scans.id).all() + + +class EmailScanConnection(relay.Connection): + class Meta: + node = EmailScan diff --git a/api/schemas/domain/email_scan/dkim/__init__.py b/api/schemas/domain/email_scan/dkim/__init__.py new file mode 100644 index 0000000000..bba8aeb2a9 --- /dev/null +++ b/api/schemas/domain/email_scan/dkim/__init__.py @@ -0,0 +1,57 @@ +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Dkim_scans +from scalars.url import URL + +from functions.get_domain import get_domain +from functions.get_timestamp import get_timestamp + +from schemas.domain.email_scan.dkim.dkim_tags import DkimTags + + +class DKIM(SQLAlchemyObjectType): + """ + DomainKeys Identified Mail (DKIM) permits a person, role, or + organization that owns the signing domain to claim some + responsibility for a message by associating the domain with the + message. This can be an author's organization, an operational relay, + or one of their agents. + """ + class Meta: + model = Dkim_scans + exclude_fields = ( + "id", + "dkim_scan" + ) + id = graphene.ID(description="ID of the object") + domain = URL(description="The domain the scan was run on") + timestamp = graphene.DateTime(description="Time when scan was initiated") + record = graphene.String( + description="DKIM record retrieved during the scan of the " + "given domain " + ) + key_length = graphene.Int( + description="Length of DKIM public key" + ) + dkim_guidance_tags = graphene.List( + lambda: DkimTags, + description="Key tags found during scan" + ) + + with app.app_context(): + def resolve_domain(self, info): + get_domain(self, info) + + def resolve_timestamp(self, info): + get_timestamp(self, info) + + def resolve_record(self, info): + return self.dkim_scan["dkim"]["txt_record"] + + def resolve_key_length(self, info): + return self.dkim_scan["dkim"]["key_size"] + + def resolve_dkim_guidance_tags(self, info): + return DkimTags.get_query(info).all() diff --git a/api/schemas/domain/email_scan/dkim/dkim_tags.py b/api/schemas/domain/email_scan/dkim/dkim_tags.py new file mode 100644 index 0000000000..9a9280a9f3 --- /dev/null +++ b/api/schemas/domain/email_scan/dkim/dkim_tags.py @@ -0,0 +1,34 @@ +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Dkim_scans + + +class DkimTags(SQLAlchemyObjectType): + class Meta: + model = Dkim_scans + exclude_fields = ( + "id", + "dkim_scan" + ) + value = graphene.String(description="Key tags found during scan") + + with app.app_context(): + def resolve_value(self: Dkim_scans, info): + tags = {} + + if "missing" in self.dkim_scan: + return tags.update({"dkim2": "missing"}) + + if self.dkim_scan["dkim"]["key_size"] >= 2048 \ + and self.dkim_scan["dkim"]["key_type"] == "rsa": + tags.update({"dkim5": "P-2048"}) + elif self.dkim_scan["dkim"]["key_size"] == 1024 \ + and self.dkim_scan["dkim"]["key_type"] == "rsa": + tags.update({"dkim4": "P-1024"}) + elif self.dkim_scan["dkim"]["key_size"] < 1024 \ + and self.dkim_scan["dkim"]["key_type"] == "rsa": + tags.update({"dkim3": "P-sub1024"}) + else: + tags.update({"dkim6": "P-invalid"}) diff --git a/api/schemas/domain/email_scan/dmarc/__init__.py b/api/schemas/domain/email_scan/dmarc/__init__.py new file mode 100644 index 0000000000..8eda3e2a2f --- /dev/null +++ b/api/schemas/domain/email_scan/dmarc/__init__.py @@ -0,0 +1,75 @@ +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Dmarc_scans + +from scalars.url import URL + +from functions.get_domain import get_domain +from functions.get_timestamp import get_timestamp + +from schemas.domain.email_scan.dmarc.dmarc_tags import DmarcTags + + +class DMARC(SQLAlchemyObjectType): + """ + Domain-based Message Authentication, Reporting, and Conformance + (DMARC) is a scalable mechanism by which a mail-originating + organization can express domain-level policies and preferences for + message validation, disposition, and reporting, that a mail-receiving + organization can use to improve mail handling. + """ + class Meta: + model = Dmarc_scans + exclude_fields = ( + "id", + "dmarc_scan" + ) + id = graphene.ID(description="ID of the object") + domain = URL(description="The domain the scan was run on") + timestamp = graphene.DateTime(description="Time when scan was initiated") + record = graphene.String( + description="DMARC record retrieved during the scan of the " + "given domain " + ) + p_policy = graphene.String( + description="The requested policy you wish mailbox providers to apply " + "when your email fails DMARC authentication and alignment" + " checks. " + ) + sp_policy = graphene.String( + description="This tag is used to indicate a requested policy for all " + "subdomains where mail is failing the DMARC " + "authentication and alignment checks. " + ) + pct = graphene.Int( + description="The percentage of messages to which the DMARC policy is " + "to be applied. " + ) + dmarc_guidance_tags = graphene.List( + lambda: DmarcTags, + description="Key tags found during DMARC Scan" + ) + + with app.app_context(): + def resolve_domain(self: Dmarc_scans, info): + return get_domain(self, info) + + def resolve_timestamp(self: Dmarc_scans, info): + return get_timestamp(self, info) + + def resolve_record(self: Dmarc_scans, info): + return self.dmarc_scan["dmarc"]["record"] + + def resolve_p_policy(self: Dmarc_scans, info): + return self.dmarc_scan["dmarc"]["tags"]["p"]["value"] + + def resolve_sp_policy(self: Dmarc_scans, info): + return self.dmarc_scan["dmarc"]["tags"]["sp"]["value"] + + def resolve_pct(self: Dmarc_scans, info): + return self.dmarc_scan["dmarc"]["tags"]["pct"]["value"] + + def resolve_dmarc_guidance_tags(self: Dmarc_scans, info): + return DmarcTags.get_query(info).all() diff --git a/api/schemas/domain/email_scan/dmarc/dmarc_tags.py b/api/schemas/domain/email_scan/dmarc/dmarc_tags.py new file mode 100644 index 0000000000..74121bf60e --- /dev/null +++ b/api/schemas/domain/email_scan/dmarc/dmarc_tags.py @@ -0,0 +1,75 @@ +import graphene +import json +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Dmarc_scans + + +class DmarcTags(SQLAlchemyObjectType): + """ + """ + class Meta: + model = Dmarc_scans + exclude_fields = ( + "id", + "dmarc_scan" + ) + + value = graphene.String(description="Important tags retrieved during scan") + + with app.app_context(): + def resolve_value(self: Dmarc_scans, info): + tags = {} + + if "missing" in self.dmarc_scan: + return tags.update({"dmarc2": "missing"}) + + # Check P Policy Tag + if self.dmarc_scan["dmarc"]["tags"]["p"]["value"] == "missing": + tags.update({"dmarc3": "P-missing"}) + elif self.dmarc_scan["dmarc"]["tags"]["p"]["value"] == "none": + tags.update({"dmarc4": "P-none"}) + elif self.dmarc_scan["dmarc"]["tags"]["p"]["value"] == "quarantine": + tags.update({"dmarc5": "P-quarantine"}) + elif self.dmarc_scan["dmarc"]["tags"]["p"]["value"] == "reject": + tags.update({"dmarc6": "P-reject"}) + + # Check PCT Tag + if self.dmarc_scan["dmarc"]["tags"]["pct"]["value"] == 100: + tags.update({"dmarc7": "PCT-100"}) + elif 100 > self.dmarc_scan["dmarc"]["tags"]["pct"]["value"] > 0: + pct_string = "PCT-" + str(self.dmarc_scan["dmarc"]["tags"]["pct"]["value"]) + tags.update({"dmarc8": pct_string}) + elif self.dmarc_scan["dmarc"]["tags"]["pct"]["value"] == "invalid": + tags.update({"dmarc9": "PCT-invalid"}) + elif self.dmarc_scan["dmarc"]["tags"]["pct"]["value"] == "none": + tags.update({"dmarc20": "PCT-none=exists"}) + else: + tags.update({"dmarc21": "PCT-0"}) + + # Check RUA Tag + for value in self.dmarc_scan["dmarc"]["tags"]["rua"]["value"]: + if value["address"] == "dmarc@cyber.gc.ca": + tags.update({"dmarc10": "RUA-CCCS"}) + else: + tags.update({"dmarc12": "RUA-none"}) + + # Check RUF Tag + for value in self.dmarc_scan["dmarc"]["tags"]["ruf"]["value"]: + if value["address"] == "dmarc@cyber.gc.ca": + tags.update({"dmarc11": "RUF-CCCS"}) + else: + tags.update({"dmarc13": "RUF-none"}) + + # Check SP tag + if self.dmarc_scan["dmarc"]["tags"]["sp"]["value"] == "missing": + tags.update({"dmarc16": "SP-missing"}) + elif self.dmarc_scan["dmarc"]["tags"]["sp"]["value"] == "none": + tags.update({"dmarc17": "SP-none"}) + elif self.dmarc_scan["dmarc"]["tags"]["sp"]["value"] == "quarantine": + tags.update({"dmarc18": "SP-quarantine"}) + elif self.dmarc_scan["dmarc"]["tags"]["sp"]["value"] == "reject": + tags.update({"dmarc19": "SP-reject"}) + + return tags diff --git a/api/schemas/domain/email_scan/spf/__init__.py b/api/schemas/domain/email_scan/spf/__init__.py new file mode 100644 index 0000000000..52b6789901 --- /dev/null +++ b/api/schemas/domain/email_scan/spf/__init__.py @@ -0,0 +1,73 @@ +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Spf_scans + +from scalars.url import URL + +from functions.get_domain import get_domain +from functions.get_timestamp import get_timestamp + +from schemas.domain.email_scan.spf.spf_tags import SPFTags + + +class SPF(SQLAlchemyObjectType): + """ + Email on the Internet can be forged in a number of ways. In + particular, existing protocols place no restriction on what a sending + host can use as the "MAIL FROM" of a message or the domain given on + the SMTP HELO/EHLO commands. Version 1 of the Sender Policy Framework (SPF) + protocol is where ADministrative Management Domains (ADMDs) can explicitly + authorize the hosts that are allowed to use their domain names, and a + receiving host can check such authorization. + """ + class Meta: + model = Spf_scans + exclude_fields = ( + "id", + "spf_scan" + ) + id = graphene.ID(description="ID of the object") + domain = URL(description="The domain the scan was run on") + timestamp = graphene.DateTime(description="The time the scan was initiated") + lookups = graphene.String( + description="The current amount of DNS lookups" + ) + record = graphene.String( + description="SPF record retrieved during the scan of the " + "given domain " + ) + spf_default = graphene.String( + description="Instruction of what a recipient should do if there is " + "not a match to your SPF record. " + ) + spf_guidance_tags = graphene.List( + lambda: SPFTags, + description="Key tags found during SPF scan" + ) + + with app.app_context(): + def resolve_domain(self: Spf_scans, info): + return get_domain(self, info) + + def resolve_timestamp(self: Spf_scans, info): + return get_timestamp(self, info) + + def resolve_lookups(self: Spf_scans, info): + return self.spf_scan["spf"]["dns_lookups"] + + def resolve_record(self: Spf_scans, info): + return self.spf_scan["spf"]["record"] + + def resolve_spf_default(self: Spf_scans, info): + if self.spf_scan["spf"]["parsed"]["all"] == "fail": + if self.spf_scan["spf"]["record"][-4:] == "-all": + return "hardfail" + elif self.spf_scan["spf"]["record"][-4:] == "~all": + return "softfail" + else: + return self.spf_scan["spf"]["parsed"]["all"] + + def resolve_spf_guidance_tags(self: Spf_scans, info): + return SPFTags.get_query(info).all() diff --git a/api/schemas/domain/email_scan/spf/spf_tags.py b/api/schemas/domain/email_scan/spf/spf_tags.py new file mode 100644 index 0000000000..995f4dad87 --- /dev/null +++ b/api/schemas/domain/email_scan/spf/spf_tags.py @@ -0,0 +1,45 @@ +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Spf_scans + + +class SPFTags(SQLAlchemyObjectType): + """ + Current settings of the currently configured SPF + """ + class Meta: + model = Spf_scans + exclude_fields = ( + "id", + "spf_scan" + ) + value = graphene.String(description="Important tags retrieved during scan") + + with app.app_context(): + def resolve_value(self: Spf_scans, info): + tags = {} + + if "missing" in self.spf_scan: + return tags.update({"spf2": "missing"}) + + # Check all tag + if self.spf_scan["spf"]["parsed"]["all"] == "missing": + tags.update({"spf3": "ALL-missing"}) + elif self.spf_scan["spf"]["parsed"]["all"] == "allow": + tags.update({"spf4": "ALL-allow"}) + elif self.spf_scan["spf"]["parsed"]["all"] == "neutral": + tags.update({"spf5": "ALL-neutral"}) + elif self.spf_scan["spf"]["parsed"]["all"] == "redirect": + tags.update({"spf8": "ALL-redirect"}) + elif self.spf_scan["spf"]["parsed"]["all"] == "fail": + if self.spf_scan["spf"]["record"][-4:] == "-all": + tags.update({"spf7": "ALL-hardfail"}) + elif self.spf_scan["spf"]["record"][-4:] == "~all": + tags.update({"spf6": "ALL-softfail"}) + + if self.spf_scan["spf"]["dns_lookups"] > 10: + tags.update({"spf10": "INCLUDE-limit"}) + + return tags diff --git a/api/schemas/domain/www_scan/__init__.py b/api/schemas/domain/www_scan/__init__.py new file mode 100644 index 0000000000..46a2c2f8d7 --- /dev/null +++ b/api/schemas/domain/www_scan/__init__.py @@ -0,0 +1,52 @@ +import graphene +from graphene import relay +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Scans, Ssl_scans, Https_scans +from scalars.url import URL + +from functions.get_domain import get_domain +from functions.get_timestamp import get_timestamp + +from schemas.domain.www_scan.https import HTTPS +from schemas.domain.www_scan.ssl import SSL + + +class WWWScan(SQLAlchemyObjectType): + """ + Results of HTTPS and SSL scans on domain + """ + class Meta: + model = Scans + interfaces = (relay.Node,) + exclude_fields = ( + "id", + "domain_id", + "scan_date", + "initiated_by" + ) + domain = URL(description="The domain the scan was run on") + timestamp = graphene.DateTime(description="The time the scan was initiated") + https = graphene.List(lambda: HTTPS) + ssl = graphene.List(lambda: SSL) + + with app.app_context(): + def resolve_domain(self: Scans, info): + return get_domain(self, info) + + def resolve_timestamp(self: Scans, info): + return get_timestamp(self, info) + + def resolve_https(self: Scans, info): + query = HTTPS.get_query(info) + return query.filter(self.id == Https_scans.id).all() + + def resolve_ssl(self: Scans, info): + query = SSL.get_query(info) + return query.filter(self.id == Ssl_scans.id).all() + + +class WWWScanConnection(relay.Connection): + class Meta: + node = WWWScan diff --git a/api/schemas/domain/www_scan/https/__init__.py b/api/schemas/domain/www_scan/https/__init__.py new file mode 100644 index 0000000000..6809a84ef3 --- /dev/null +++ b/api/schemas/domain/www_scan/https/__init__.py @@ -0,0 +1,54 @@ +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Https_scans +from scalars.url import URL + +from functions.get_domain import get_domain +from functions.get_timestamp import get_timestamp + +from schemas.domain.www_scan.https.https_tags import HTTPSTags + + +class HTTPS(SQLAlchemyObjectType): + class Meta: + model = Https_scans + exclude_fields = ( + "id", + "https_scan" + ) + id = graphene.ID(description="The ID of the object") + domain = URL(description="The domain the scan was run on") + timestamp = graphene.DateTime(description="The time the scan was initiated") + implementation = graphene.String() + enforced = graphene.String() + hsts = graphene.String() + hsts_age = graphene.String() + preloaded = graphene.String() + https_guidance_tags = graphene.List(lambda: HTTPSTags) + + with app.app_context(): + def resole_domain(self: Https_scans, info): + return get_domain(self, info) + + def resolve_timestamp(self: Https_scans, info): + return get_timestamp(self, info) + + def resolve_implementation(self: Https_scans, info): + return self.https_scan["https"]["implementation"] + + def resolve_enforced(self: Https_scans, info): + return self.https_scan["https"]["enforced"] + + def resolve_hsts(self: Https_scans, info): + return self.https_scan["https"]["hsts"] + + def resolve_hsts_age(self: Https_scans, info): + return self.https_scan["https"]["hsts_age"] + + def resolve_preloaded(self: Https_scans, info): + return self.https_scan["https"]["preloaded"] + + def resolve_https_guidance_tags(self: Https_scans, info): + return HTTPS.get_query(info).all() diff --git a/api/schemas/domain/www_scan/https/https_tags.py b/api/schemas/domain/www_scan/https/https_tags.py new file mode 100644 index 0000000000..51d13d45a6 --- /dev/null +++ b/api/schemas/domain/www_scan/https/https_tags.py @@ -0,0 +1,20 @@ +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Https_scans + + +class HTTPSTags(SQLAlchemyObjectType): + class Meta: + model = Https_scans + exclude_fields = ( + "id", + "https_scan" + ) + value = graphene.String() + + with app.app_context(): + def resolve_value(self, info): + tags = {} + return tags diff --git a/api/schemas/domain/www_scan/ssl/__init__.py b/api/schemas/domain/www_scan/ssl/__init__.py new file mode 100644 index 0000000000..4b07f12bf7 --- /dev/null +++ b/api/schemas/domain/www_scan/ssl/__init__.py @@ -0,0 +1,29 @@ +import graphene +from graphene_sqlalchemy import SQLAlchemyObjectType + +from app import app +from models import Ssl_scans + +from scalars.url import URL + +from functions.get_domain import get_domain +from functions.get_timestamp import get_timestamp + + +class SSL(SQLAlchemyObjectType): + class Meta: + model = Ssl_scans + exclude_fields = ( + "id", + "ssl_scan" + ) + id = graphene.ID() + domain = URL() + timestamp = graphene.DateTime() + + with app.app_context(): + def resolve_domain(self, info): + return get_domain(self, info) + + def resolve_timestamp(self, info): + return get_timestamp(self, info) diff --git a/api/schemas/domain/www_scan/ssl/ssl_tags.py b/api/schemas/domain/www_scan/ssl/ssl_tags.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/api/schemas/domains.py b/api/schemas/domains.py deleted file mode 100644 index 88bae5e732..0000000000 --- a/api/schemas/domains.py +++ /dev/null @@ -1,16 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField - -from models import Domains as DomainModel - - -class Domains(SQLAlchemyObjectType): - class Meta: - model = DomainModel - interfaces = (relay.Node,) - - -class DomainsConnection(relay.Connection): - class Meta: - node = Domains diff --git a/api/schemas/groups.py b/api/schemas/groups.py deleted file mode 100644 index c8c839de36..0000000000 --- a/api/schemas/groups.py +++ /dev/null @@ -1,16 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType - -from models import Groups as GroupsModel - - -class Groups(SQLAlchemyObjectType): - class Meta: - model = GroupsModel - interfaces = (relay.Node,) - - -class GroupsConnection(relay.Connection): - class Meta: - node = Groups diff --git a/api/schemas/http.py b/api/schemas/http.py deleted file mode 100644 index e0db681b47..0000000000 --- a/api/schemas/http.py +++ /dev/null @@ -1,16 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType - -from models import Https_scans as HttpModel - - -class HTTP(SQLAlchemyObjectType): - class Meta: - model = HttpModel - interfaces = (relay.Node,) - - -class HttpConnection(relay.Connection): - class Meta: - node = HTTP diff --git a/api/schemas/organizations.py b/api/schemas/organizations.py index 679a6f68cd..0d5b1489dc 100644 --- a/api/schemas/organizations.py +++ b/api/schemas/organizations.py @@ -1,16 +1,15 @@ -import graphene from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField +from graphene_sqlalchemy import SQLAlchemyObjectType -from models import Organizations as OrganizationsModel +from models import Organizations as OrgModel -class Organizations(SQLAlchemyObjectType): +class Organization(SQLAlchemyObjectType): class Meta: - model = OrganizationsModel - interfaces = (relay.Node,) + model = OrgModel + interface = (relay.Node, ) -class OrganizationsConnection(relay.Connection): +class OrganizationConnection(relay.Connection): class Meta: - node = Organizations + node = Organization diff --git a/api/schemas/scans.py b/api/schemas/scans.py deleted file mode 100644 index f3bf7a8ce4..0000000000 --- a/api/schemas/scans.py +++ /dev/null @@ -1,16 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType - -from models import Scans as ScanModel - - -class Scans(SQLAlchemyObjectType): - class Meta: - model = ScanModel - interfaces = (relay.Node,) - - -class ScansConnection(relay.Connection): - class Meta: - node = Scans diff --git a/api/schemas/sectors.py b/api/schemas/sectors.py deleted file mode 100644 index ca94911897..0000000000 --- a/api/schemas/sectors.py +++ /dev/null @@ -1,17 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField -from graphene_sqlalchemy.enums import * - -from models import Sectors as SectorsModel - - -class Sectors(SQLAlchemyObjectType): - class Meta: - model = SectorsModel - interfaces = (relay.Node,) - - -class SectorsConnection(relay.Connection): - class Meta: - node = Sectors diff --git a/api/schemas/spf.py b/api/schemas/spf.py deleted file mode 100644 index 64b5640cfa..0000000000 --- a/api/schemas/spf.py +++ /dev/null @@ -1,16 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType - -from models import Spf_scans as SpfModel - - -class Spf(SQLAlchemyObjectType): - class Meta: - model = SpfModel - interfaces = (relay.Node,) - - -class SpfConnection(relay.Connection): - class Meta: - node = Spf diff --git a/api/schemas/ssl.py b/api/schemas/ssl.py deleted file mode 100644 index 45b198ccf1..0000000000 --- a/api/schemas/ssl.py +++ /dev/null @@ -1,16 +0,0 @@ -import graphene -from graphene import relay -from graphene_sqlalchemy import SQLAlchemyObjectType - -from models import Ssl_scans as SSLModel - - -class SSL(SQLAlchemyObjectType): - class Meta: - model = SSLModel - interfaces = (relay.Node,) - - -class SSLConnection(relay.Connection): - class Meta: - node = SSL diff --git a/api/tests/test_cost_check.py b/api/tests/test_cost_check.py index 93ef8eb946..699ad0621b 100644 --- a/api/tests/test_cost_check.py +++ b/api/tests/test_cost_check.py @@ -17,7 +17,7 @@ from db import db from app import app from queries import schema -from models import Sectors, Groups +from models import Users from backend.security_check import SecurityAnalysisBackend remove_seed() @@ -27,25 +27,17 @@ def user_schema_test_db_init(): db.init_app(app) with app.app_context(): - test_sector = Sectors( + test_user = Users( id=1, - zone='ZO1', - sector='SO1' + display_name="test" ) - db.session.add(test_sector) - test_group = Groups( - id=1, - s_group='GO1', - sector_id=1 - ) - db.session.add(test_group) + db.session.add(test_user) db.session.commit() yield with app.app_context(): - Groups.query.delete() - Sectors.query.delete() + Users.query.delete() db.session.commit() @@ -54,40 +46,35 @@ def user_schema_test_db_init(): @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 { + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + query = client.execute( + ''' + { + users { edges { node { - sectorId + displayName } } } } - } - ''', backend=backend) - result_refr = { - "data": { - "getSectorById": [ - { - "groups": { - "edges": [ - { - "node": { - "sectorId": 1 - } + ''', backend=backend) + result_refr = { + "data": { + "users": { + "edges": [ + { + "node": { + "displayName": "test" } - ] - } + } + ] } - ] + } } - } - self.assertDictEqual(result_refr, query) + self.assertDictEqual(result_refr, query) def test_invalid_cost_query(self): backend = SecurityAnalysisBackend(10, 5) diff --git a/api/tests/test_depth_check.py b/api/tests/test_depth_check.py index ad946056b8..5c60d61fe8 100644 --- a/api/tests/test_depth_check.py +++ b/api/tests/test_depth_check.py @@ -17,7 +17,7 @@ from app import app from db import db from queries import schema -from models import Sectors, Groups +from models import Users from backend.security_check import SecurityAnalysisBackend remove_seed() @@ -27,25 +27,17 @@ def user_schema_test_db_init(): db.init_app(app) with app.app_context(): - test_sector = Sectors( + test_user = Users( id=1, - zone='ZO1', - sector='SO1' + display_name="test" ) - db.session.add(test_sector) - test_group = Groups( - id=1, - s_group='GO1', - sector_id=1 - ) - db.session.add(test_group) + db.session.add(test_user) db.session.commit() yield with app.app_context(): - Groups.query.delete() - Sectors.query.delete() + Users.query.delete() db.session.commit() @@ -54,40 +46,35 @@ def user_schema_test_db_init(): @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 { + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + query = client.execute( + ''' + { + users { edges { node { - sectorId + displayName } } } } - } - ''', backend=backend) - result_refr = { - "data": { - "getSectorById": [ - { - "groups": { - "edges": [ - { - "node": { - "sectorId": 1 - } + ''', backend=backend) + result_refr = { + "data": { + "users": { + "edges": [ + { + "node": { + "displayName": "test" } - ] - } + } + ] } - ] + } } - } - self.assertDictEqual(result_refr, query) + self.assertDictEqual(result_refr, query) def test_invalid_depth_query(self): backend = SecurityAnalysisBackend(10) diff --git a/api/tests/test_domains_resolver.py b/api/tests/test_domains_resolver.py deleted file mode 100644 index 82b683f49f..0000000000 --- a/api/tests/test_domains_resolver.py +++ /dev/null @@ -1,179 +0,0 @@ -import sys -import os -from os.path import dirname, join, expanduser, normpath, realpath - -import pytest -from graphene.test import Client - -from unittest import TestCase - -from manage import seed, remove_seed - -seed() -from app import app -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. -PACKAGE_PARENT = '..' -SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__)))) -sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT))) - - -@pytest.fixture(scope='class') -def domain_test_db_init(): - db.init_app(app) - with app.app_context(): - org = Organizations( - id=2, - organization='ORG2', - description='Organization 2', - ) - db.session.add(org) - db.session.commit() - - domain = Domains( - id=1, - domain='somecooldomain.ca', - organization_id=2 - ) - db.session.add(domain) - db.session.commit() - - yield - - with app.app_context(): - Domains.query.delete() - Organizations.query.delete() - db.session.commit() - - -@pytest.mark.usefixtures('domain_test_db_init') -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 = """ - { - getDomainById(id: 1){ - domain - } - }""" - result_refr = { - "data": { - "getDomainById": [ - { - "domain": "somecooldomain.ca" - } - ] - } - } - - 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 = """ - { - getDomainByDomain(url: "somecooldomain.ca"){ - domain - } - }""" - result_refr = { - "data": { - "getDomainByDomain": [ - { - "domain": "somecooldomain.ca" - } - ] - } - } - - 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 = """ - { - getDomainByOrganization(org: ORG2){ - domain - } - }""" - result_refr = { - "data": { - "getDomainByOrganization": [ - { - "domain": "somecooldomain.ca" - } - ] - } - } - - 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 = """ - { - getDomainById(id: 9999){ - domain - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, Invalid ID" - - 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 = """ - { - getDomainByDomain(url: "google.ca"){ - domain - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0][ - 'message'] == 'Error, domain does not exist' - - 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 = """ - { - getDomainByOrganization(org: fds){ - domain - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0][ - 'message'] == f'Argument "org" has invalid value fds.\nExpected type "OrganizationsEnum", found fds.' diff --git a/api/tests/test_domains_resolver_access_control.py b/api/tests/test_domains_resolver_access_control.py new file mode 100644 index 0000000000..8c35a44166 --- /dev/null +++ b/api/tests/test_domains_resolver_access_control.py @@ -0,0 +1,638 @@ +import sys +import os +from os.path import dirname, join, expanduser, normpath, realpath + +import pytest +from flask import Request +from graphene.test import Client +from flask_bcrypt import Bcrypt + +from unittest import TestCase + +from werkzeug.test import create_environ + +from manage import seed, remove_seed + +seed() +from app import app +from db import db +from models import Organizations, Domains, Users, User_affiliations +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. +PACKAGE_PARENT = '..' +SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__)))) +sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT))) + + +@pytest.fixture(scope='class') +def domain_test_db_init(): + db.init_app(app) + bcrypt = Bcrypt(app) + + with app.app_context(): + test_user = Users( + id=1, + display_name="testuserread", + user_name="testuserread@testemail.ca", + user_password=bcrypt.generate_password_hash( + password="testpassword123").decode("UTF-8"), + ) + db.session.add(test_user) + test_super_admin = Users( + id=2, + display_name="testsuperadmin", + user_name="testsuperadmin@testemail.ca", + user_password=bcrypt.generate_password_hash( + password="testpassword123").decode("UTF-8") + ) + db.session.add(test_super_admin) + + org = Organizations( + id=1, + organization='ORG1', + description='Organization 1', + ) + db.session.add(org) + org = Organizations( + id=2, + organization='ORG2', + description='Organization 2', + ) + db.session.add(org) + org = Organizations( + id=3, + organization='ORG3', + description='Organization 3', + ) + db.session.add(org) + + test_admin_role = User_affiliations( + user_id=1, + organization_id=1, + permission='user_read' + ) + db.session.add(test_admin_role) + test_admin_role = User_affiliations( + user_id=2, + organization_id=1, + permission='super_admin' + ) + db.session.add(test_admin_role) + + domain = Domains( + id=1, + domain='somecooldomain.ca', + organization_id=1 + ) + db.session.add(domain) + domain = Domains( + id=2, + domain='anothercooldomain.ca', + organization_id=1 + ) + db.session.add(domain) + domain = Domains( + id=3, + domain='somelamedomain.ca', + organization_id=2 + ) + db.session.add(domain) + db.session.commit() + + yield + + with app.app_context(): + Domains.query.delete() + User_affiliations.query.delete() + Organizations.query.delete() + Users.query.delete() + db.session.commit() + + +@pytest.mark.usefixtures('domain_test_db_init') +class TestDomainsResolver(TestCase): + # Super Admin Tests + def test_get_domain_resolvers_by_url_super_admin_single_node(self): + """Test domain resolver by url as a super admin, single node return""" + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testsuperadmin@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domain(url: "somelamedomain.ca") { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + result_refr = { + "data": { + "domain": { + "edges": [ + { + "node": { + "url": "somelamedomain.ca" + } + } + ] + } + } + } + self.assertDictEqual(result_refr, executed) + + def test_get_domain_resolvers_by_org_super_admin_single_node(self): + """ + Test domain resolver by organization as a super admin, single node + return + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testsuperadmin@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domains(organization: ORG2) { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + result_refr = { + "data": { + "domains": { + "edges": [ + { + "node": { + "url": "somelamedomain.ca" + } + } + ] + } + } + } + self.assertDictEqual(result_refr, executed) + + def test_get_domain_resolvers_by_org_super_admin_multi_node(self): + """ + Test domain resolver by organization as a super admin, multi node return + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testsuperadmin@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domains(organization: ORG1) { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + result_refr = { + "data": { + "domains": { + "edges": [ + { + "node": { + "url": "somecooldomain.ca" + } + }, + { + "node": { + "url": "anothercooldomain.ca" + } + } + ] + } + } + } + self.assertDictEqual(result_refr, executed) + + def test_get_domain_resolvers_by_url_super_admin_invalid_domain(self): + """ + Test domain resolver by url as a super admin, invalid domain + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testsuperadmin@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domain(url: "google.ca") { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + assert executed['errors'] + assert executed['errors'][0] + assert executed['errors'][0]['message'] == "Error, domain does not exist" + + def test_get_domain_resolvers_by_org_super_admin_org_no_domains(self): + """ + Test domain resolver by org as a super admin, org has no domains + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testsuperadmin@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domains(organization: ORG3) { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + assert executed['errors'] + assert executed['errors'][0] + assert executed['errors'][0]['message'] == "Error, no domains associated with that organization" + + # User read tests + def test_get_domain_resolvers_by_url_user_read_single_node(self): + """ + Test domain resolver get domain by url as user read, return as + single node + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testuserread@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domain(url: "somecooldomain.ca") { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + result_refr = { + "data": { + "domain": { + "edges": [ + { + "node": { + "url": "somecooldomain.ca" + } + } + ] + } + } + } + self.assertDictEqual(result_refr, executed) + + def test_get_domain_resolvers_by_org_user_read_multi_node(self): + """ + Test domain resolver get domain by org as user read, return as + multi node + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testuserread@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domains(organization: ORG1) { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + result_refr = { + "data": { + "domains": { + "edges": [ + { + "node": { + "url": "somecooldomain.ca" + } + }, + { + "node": { + "url": "anothercooldomain.ca" + } + } + ] + } + } + } + self.assertDictEqual(result_refr, executed) + + def test_get_domain_resolvers_by_url_user_read_no_access(self): + """ + Test domain resolver get domain by url as user read, user has no rights + to view domains related to that org + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testuserread@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domain(url: "somelamedomain.ca") { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + assert executed['errors'] + assert executed['errors'][0] + assert executed['errors'][0]['message'] == "Error, you do not have permission to view that domain" + + def test_get_domain_resolvers_by_org_user_read_no_access(self): + """ + Test domain resolver get domain by org as user read, user has no rights + to view domains related to that org + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testuserread@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domains(organization: ORG2) { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + assert executed['errors'] + assert executed['errors'][0] + assert executed['errors'][0]['message'] == "Error, you do not have permission to view that organization" + + def test_get_domain_resolvers_by_url_user_read_invalid_domain(self): + """ + Test domain resolver get domain by url as user read, url does not + exist + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testuserread@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domain(url: "google.ca") { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + assert executed['errors'] + assert executed['errors'][0] + assert executed['errors'][0]['message'] == "Error, domain does not exist" + + def test_get_domain_resolvers_by_org_user_read_org_no_domains(self): + """ + Test domain resolver get domain by org as user read, org has no related + domains + """ + with app.app_context(): + backend = SecurityAnalysisBackend() + client = Client(schema) + get_token = client.execute( + ''' + mutation{ + signIn(userName:"testuserread@testemail.ca", password:"testpassword123"){ + authToken + } + } + ''', backend=backend) + assert get_token['data']['signIn']['authToken'] is not None + token = get_token['data']['signIn']['authToken'] + assert token is not None + + environ = create_environ() + environ.update( + HTTP_AUTHORIZATION=token + ) + request_headers = Request(environ) + + executed = client.execute( + ''' + { + domains(organization: ORG3) { + edges { + node { + url + } + } + } + } + ''', context_value=request_headers, backend=backend) + assert executed['errors'] + assert executed['errors'][0] + assert executed['errors'][0]['message'] == "Error, you do not have permission to view that organization" diff --git a/api/tests/test_groups_resolver.py b/api/tests/test_groups_resolver.py deleted file mode 100644 index 21bc8f1920..0000000000 --- a/api/tests/test_groups_resolver.py +++ /dev/null @@ -1,217 +0,0 @@ -import sys -import os -from os.path import dirname, join, expanduser, normpath, realpath - -import pytest -from graphene.test import Client - -from unittest import TestCase - -from manage import seed, remove_seed - -seed() -from app import app -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. -PACKAGE_PARENT = '..' -SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__)))) -sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT))) - - -@pytest.fixture(scope='class') -def group_test_db_init(): - db.init_app(app) - with app.app_context(): - sector = Sectors( - id=1, - zone="ZO1", - sector="SEC1", - description="Sector 1" - ) - db.session.add(sector) - sector = Sectors( - id=2, - zone="ZO2", - sector="SEC2", - description="Sector 2" - ) - db.session.add(sector) - sector = Sectors( - id=25, - zone="TEST", - sector="TEST_DEV", - description="Development test cases" - ) - db.session.add(sector) - db.session.commit() - group = Groups( - id=1, - s_group='GO1', - description='Group 1', - sector_id=1 - ) - db.session.add(group) - group = Groups( - id=2, - s_group='GO2', - description='Group 2', - sector_id=2 - ) - db.session.add(group) - - yield - - with app.app_context(): - Groups.query.delete() - Sectors.query.delete() - db.session.commit() - -@pytest.mark.usefixtures('group_test_db_init') -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 = """ - { - getGroupById(id:1) { - sGroup, - description - } - }""" - result_refr = { - "data": { - "getGroupById": [ - { - "sGroup": "GO1", - "description": "Group 1" - } - ] - } - } - - 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 = """ - { - getGroupByGroup(group: GO1){ - description - sectorId - } - }""" - result_refr = { - "data": { - "getGroupByGroup": [ - { - "description": "Group 1", - "sectorId": 1 - } - ] - } - } - - 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 = """ - { - getGroupBySector(sector: SEC1){ - description - groupSector{ - zone - description - } - } - }""" - result_refr = { - "data": { - "getGroupBySector": [ - { - "description": "Group 1", - "groupSector": { - "zone": "ZO1", - "description": "Sector 1" - } - } - ] - } - } - - 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 = """ - { - getGroupById(id: 9999){ - id - description - sectorId - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, Invalid ID" - - 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 = """ - { - getGroupByGroup(group: fds){ - id - description - sectorId - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0][ - 'message'] == f'Argument "group" has invalid value fds.\nExpected type "GroupEnums", found fds.' - - 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 = """ - { - getGroupBySector(sector: dsa){ - id - description - sectorId - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0][ - 'message'] == f'Argument "sector" has invalid value dsa.\nExpected type "SectorEnums", found dsa.' diff --git a/api/tests/test_organizations_resolver.py b/api/tests/test_organizations_resolver.py deleted file mode 100644 index ab2408b850..0000000000 --- a/api/tests/test_organizations_resolver.py +++ /dev/null @@ -1,188 +0,0 @@ -import sys -import json -import os -from os.path import dirname, join, expanduser, normpath, realpath - -import pytest -from graphene.test import Client - -from unittest import TestCase - -from manage import seed, remove_seed - -seed() -from app import app -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. -PACKAGE_PARENT = '..' -SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__)))) -sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT))) - - -@pytest.fixture(scope='class') -def org_test_db_build(): - db.init_app(app) - - with app.app_context(): - group = Groups( - id=1, - s_group='GO1', - description='Group 1', - ) - db.session.add(group) - - org = Organizations( - id=1, - organization='ORG1', - description='Organization 1', - group_id=1 - ) - db.session.add(org) - db.session.commit() - - yield - - with app.app_context(): - Organizations.query.delete() - Groups.query.delete() - db.session.commit() - - -@pytest.mark.usefixtures("org_test_db_build") -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 = """ - { - getOrgById(id: 1){ - description - organization - } - }""" - result_refr = { - "data": { - "getOrgById": [ - { - "description": "Organization 1", - "organization": "ORG1" - } - ] - } - } - - 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 = """ - { - getOrgByOrg(org: ORG1){ - description - organization - } - }""" - result_refr = { - "data": { - "getOrgByOrg": [ - { - "description": "Organization 1", - "organization": "ORG1" - } - ] - } - } - - 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 = """ - { - getOrgByGroup(group: GO1){ - description - groupId - } - }""" - result_refr = { - "data": { - "getOrgByGroup": [ - { - "description": "Organization 1", - "groupId": 1 - } - ] - } - } - - 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 = """ - { - getOrgById(id: 9999){ - description - groupId - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, Invalid ID" - - 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 = """ - { - getOrgByOrg(org: fds){ - id - description - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == f'Argument "org" has invalid value fds.\nExpected type "OrganizationsEnum", found fds.' - - 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 = """ - { - getOrgByGroup(group: dsa){ - id - description - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == f'Argument "group" has invalid value dsa.\nExpected type "GroupEnums", found dsa.' diff --git a/api/tests/test_scans_resolver.py b/api/tests/test_scans_resolver.py deleted file mode 100644 index 1819ab2013..0000000000 --- a/api/tests/test_scans_resolver.py +++ /dev/null @@ -1,390 +0,0 @@ -import sys -import os -from os.path import dirname, join, expanduser, normpath, realpath - -import pytest -from graphene.test import Client - -from flask_bcrypt import Bcrypt - -from unittest import TestCase - -from manage import seed, remove_seed - -seed() -from app import app -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. -PACKAGE_PARENT = '..' -SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__)))) -sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT))) - - -@pytest.fixture(scope='class') -def scans_test_db_init(): - db.init_app(app) - with app.app_context(): - bcrypt = Bcrypt(app) - user = Users( - id=1, - display_name="testuser", - user_name="testuser@testemail.ca", - user_password=bcrypt.generate_password_hash( - password="testpassword123").decode("UTF-8") - ) - db.session.add(user) - user = Users( - id=2 - ) - db.session.add(user) - domain = Domains( - id=1, - domain='valid.canada.ca' - ) - db.session.add(domain) - domain = Domains( - id=2, - domain='www.testdomain.ca' - ) - db.session.add(domain) - scan = Scans( - id=1, - scan_date="2020-02-18T09:43:14", - domain_id=1, - initiated_by=1 - ) - db.session.add(scan) - scan = Scans( - id=2, - scan_date="2020-02-15T09:43:17", - domain_id=1, - initiated_by=1 - ) - db.session.add(scan) - db.session.commit() - - yield - - with app.app_context(): - Scans.query.delete() - Domains.query.delete() - Users.query.delete() - db.session.commit() - - -@pytest.mark.usefixtures('scans_test_db_init') -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 = """ - { - getScanById(id: 1) { - scanDate - domain { - domain - } - } - } - """ - - result_refr = { - "data": { - "getScanById": [ - { - "scanDate": "2020-02-18T09:43:14", - "domain": { - "domain": "valid.canada.ca" - } - } - ] - } - } - - 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 = """ - { - getScansByDate(date: "2020-02-18") { - scanDate - domain { - domain - } - } - } - """ - - result_refr = { - "data": { - "getScansByDate": [ - { - "scanDate": "2020-02-18T09:43:14", - "domain": { - "domain": "valid.canada.ca" - } - } - ] - } - } - - 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 = """ - { - getScansByDateRange(startDate: "2020-02-15", endDate: "2020-02-18"){ - scanDate - domain { - domain - } - } - } - """ - - result_refr = { - "data": { - "getScansByDateRange": [ - { - "scanDate": "2020-02-18T09:43:14", - "domain": { - "domain": "valid.canada.ca" - } - }, - { - "scanDate": "2020-02-15T09:43:17", - "domain": { - "domain": "valid.canada.ca" - } - } - ] - } - } - - 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 = """ - { - getScansByDomain(url: "valid.canada.ca") { - scanDate - domain { - domain - } - } - } - """ - - result_refr = { - "data": { - "getScansByDomain": [ - { - "scanDate": "2020-02-18T09:43:14", - "domain": { - "domain": "valid.canada.ca" - } - }, - { - "scanDate": "2020-02-15T09:43:17", - "domain": { - "domain": "valid.canada.ca" - } - } - ] - } - } - - 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 = """ - { - getScansByUserId(id: 1) { - initiatedBy - domain { - domain - } - } - } - """ - - result_refr = { - "data": { - "getScansByUserId": [ - { - "initiatedBy": 1, - "domain": { - "domain": "valid.canada.ca" - } - }, - { - "initiatedBy": 1, - "domain": { - "domain": "valid.canada.ca" - } - } - ] - } - } - - 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 = """ - { - getScanById(id: 9999){ - id - domainId - } - } - """ - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, Invalid ID" - - 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 = """ - { - getScansByDate(date: "1970-01-01"){ - id - } - } - """ - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, No scans occurred on that date" - - 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 = """ - { - getScansByDateRange(startDate: "1970-01-01", endDate: "1980-01-01") { - id - } - } - """ - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, No scans in that date range" - - 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 = """ - { - getScansByDomain(url: "www.testfakedomain.ca") { - id - } - } - """ - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, no domain associated with that URL" - - 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 = """ - { - getScansByDomain(url: "www.testdomain.ca") { - id - } - } - """ - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, no scans associated with that domain" - - 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 = """ - { - getScansByUserId(id: 2) { - id - } - } - """ - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, no scans initiated by that user" - - 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 = """ - { - getScansByUserId(id: 999) { - id - } - } - """ - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, cannot find user" diff --git a/api/tests/test_sector_resolver.py b/api/tests/test_sector_resolver.py deleted file mode 100644 index 32649c814e..0000000000 --- a/api/tests/test_sector_resolver.py +++ /dev/null @@ -1,206 +0,0 @@ -import sys -import os -from os.path import dirname, join, expanduser, normpath, realpath - -import pytest -from graphene.test import Client - -from unittest import TestCase - -from manage import seed, remove_seed - -seed() -from app import app -from db import db -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. -PACKAGE_PARENT = '..' -SCRIPT_DIR = dirname(realpath(join(os.getcwd(), expanduser(__file__)))) -sys.path.append(normpath(join(SCRIPT_DIR, PACKAGE_PARENT))) - - -@pytest.fixture(scope='class') -def sector_test_db_init(): - db.init_app(app) - with app.app_context(): - sector = Sectors( - id=1, - zone="ZO1", - sector="SEC1", - description="Sector 1" - ) - with app.app_context(): - db.session.add(sector) - db.session.commit() - - sector = Sectors( - id=2, - zone="ZO2", - sector="SEC2", - description="Sector 2" - ) - db.session.add(sector) - - sector = Sectors( - id=25, - zone="TEST", - sector="TEST_DEV", - description="Development test cases" - ) - db.session.add(sector) - db.session.commit() - - yield - - with app.app_context(): - Sectors.query.delete() - db.session.commit() - - -@pytest.mark.usefixtures('sector_test_db_init') -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 = """ - { - getSectorById(id: 1) { - sector - zone - description - } - }""" - result_refr = { - "data": { - "getSectorById": [ - { - "sector": "SEC1", - "zone": "ZO1", - "description": "Sector 1" - } - ] - } - } - - 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 = """ - { - getSectorsBySector(sector: SEC1){ - zone - description - } - }""" - result_refr = { - "data": { - "getSectorsBySector": [ - { - "zone": "ZO1", - "description": "Sector 1" - } - ] - } - } - - 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 = """ - { - getSectorByZone(zone: TEST) { - sector - description - } - }""" - result_refr = { - "data": { - "getSectorByZone": [ - { - "sector": "TEST_DEV", - "description": "Development test cases" - } - ] - } - } - - 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 = """ - { - getSectorById(id: 9999){ - id - sector - zone - description - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == "Error, Invalid ID" - - 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 = """ - { - getSectorsBySector(sector: str) { - id - zone - description - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == f'Argument "sector" has invalid value str.\nExpected type "SectorEnums", found str.' - - 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 = """ - { - getSectorByZone(zone: str) { - id - sector - zone - description - } - }""" - executed = client.execute(query, backend=backend) - - assert executed['errors'] - assert executed['errors'][0] - assert executed['errors'][0]['message'] == f'Argument "zone" has invalid value str.\nExpected type "ZoneEnums", found str.'