test_api_workspace.py

'''
Faraday Penetration Test IDE
Copyright (C) 2013  Infobyte LLC (http://www.infobytesec.com/)
See the file 'doc/LICENSE' for the license information

'''
from datetime import date
import time
from urllib.parse import urljoin

import pytest
from posixpath import join

from faraday.server.models import Workspace, Scope, SeveritiesHistogram
from faraday.server.api.modules.workspaces import WorkspaceView
from tests.test_api_non_workspaced_base import ReadWriteAPITests, BulkDeleteTestsMixin
from tests import factories


class TestWorkspaceAPI(ReadWriteAPITests, BulkDeleteTestsMixin):
    model = Workspace
    factory = factories.WorkspaceFactory
    api_endpoint = 'ws'
    lookup_field = 'name'
    view_class = WorkspaceView
    patchable_fields = ['description']

    @pytest.mark.usefixtures('ignore_nplusone')
    def test_filter_restless_by_name(self, test_client):
        res = test_client.get(
            join(
                self.url(),
                f'filter?q={{"filters":[{{"name": "name", "op":"eq", "val": "{self.first_object.name}"}}]}}'
            )
        )
        assert res.status_code == 200
        assert len(res.json) == 1
        assert res.json[0]['name'] == self.first_object.name

    @pytest.mark.usefixtures('ignore_nplusone')
    def test_filter_restless_by_name_zero_results_found(self, test_client):
        res = test_client.get(
            join(
                self.url(),
                'filter?q={"filters":[{"name": "name", "op":"eq", "val": "thiswsdoesnotexist"}]}'
            )
        )
        assert res.status_code == 200
        assert len(res.json) == 0

    def test_filter_restless_by_description(self, test_client):
        self.first_object.description = "this is a new description"
        res = test_client.get(
            join(
                self.url(),
                f'filter?q={{"filters":[{{"name": "description", "op":"eq", "val": "{self.first_object.description}"}}'
                ']}'
            )
        )
        assert res.status_code == 200
        assert len(res.json) == 1
        assert res.json[0]['description'] == self.first_object.description

    def test_filter_restless_with_vulns_stats(self, test_client, vulnerability_factory,
                                              vulnerability_web_factory, session):

        vulns = vulnerability_factory.create_batch(8, workspace=self.first_object,
                                                   confirmed=False, status='open', severity='informational')

        vulns += vulnerability_factory.create_batch(3, workspace=self.first_object,
                                                    confirmed=True, status='closed', severity='critical')

        vulns += vulnerability_web_factory.create_batch(2, workspace=self.first_object,
                                                    confirmed=True, status='open', severity='low')

        vulns += vulnerability_web_factory.create_batch(2, workspace=self.first_object,
                                                        confirmed=True, status='open', severity='high')

        vulns += vulnerability_web_factory.create_batch(2, workspace=self.first_object,
                                                        confirmed=True, status='open', severity='unclassified')

        vulns += vulnerability_web_factory.create_batch(2, workspace=self.first_object,
                                                        confirmed=True, status='open', severity='medium')

        session.add_all(vulns)
        session.commit()

        self.first_object.description = "this is a new description"
        res = test_client.get(
            join(
                self.url(),
                f'filter?q={{"filters":[{{"name": "description", "op":"eq", "val": "{self.first_object.description}"}}'
                ']}'
            )
        )
        assert res.status_code == 200
        assert len(res.json) == 1

        assert res.json[0]['stats']['std_vulns'] == 11
        assert res.json[0]['stats']['web_vulns'] == 8
        assert res.json[0]['stats']['code_vulns'] == 0
        assert res.json[0]['active_agents_count'] == 0

        assert res.json[0]['description'] == self.first_object.description
        assert res.json[0]['stats']['total_vulns'] == 19
        assert res.json[0]['stats']['info_vulns'] == 8
        assert res.json[0]['stats']['critical_vulns'] == 3
        assert res.json[0]['stats']['low_vulns'] == 2
        assert res.json[0]['stats']['high_vulns'] == 2
        assert res.json[0]['stats']['medium_vulns'] == 2
        assert res.json[0]['stats']['unclassified_vulns'] == 2

    def test_host_count(self, host_factory, test_client, session):
        host_factory.create(workspace=self.first_object)
        session.commit()
        res = test_client.get(self.url(self.first_object))
        assert res.status_code == 200
        assert res.json['stats']['hosts'] == 1

    @pytest.mark.parametrize('querystring', [
        '?confirmed=1&status=open',
        '?confirmed=true&status=open'
    ])
    def test_vuln_count_open_confirmed(self,
                        vulnerability_factory,
                        vulnerability_web_factory,
                        test_client,
                        session,
                        querystring):
        vulns = vulnerability_factory.create_batch(8, workspace=self.first_object,
                                                   confirmed=False, status='open', severity='critical')

        vulns += vulnerability_factory.create_batch(3, workspace=self.first_object,
                                                    confirmed=True, status='closed', severity='critical')

        vulns += vulnerability_web_factory.create_batch(2, workspace=self.first_object,
                                                    confirmed=True, status='open', severity='informational')

        session.add_all(vulns)
        session.commit()
        res = test_client.get(urljoin(self.url(self.first_object), querystring))
        assert res.status_code == 200
        assert res.json['stats']['code_vulns'] == 0
        assert res.json['stats']['web_vulns'] == 2
        assert res.json['stats']['std_vulns'] == 0
        assert res.json['stats']['critical_vulns'] == 0
        assert res.json['stats']['info_vulns'] == 2
        assert res.json['stats']['total_vulns'] == 2
        assert res.json['last_run_agent_date'] is None
        assert res.json['stats']['opened_vulns'] == 10
        assert res.json['stats']['confirmed_vulns'] == 2

    @pytest.mark.skip_sql_dialect('sqlite')
    @pytest.mark.usefixtures('ignore_nplusone')
    def test_histogram(self,
                        vulnerability_factory,
                        vulnerability_web_factory,
                        second_workspace,
                        test_client,
                        session):

        session.query(SeveritiesHistogram).delete()
        session.commit()

        vulns = vulnerability_factory.create_batch(8, workspace=self.first_object,
                                                   confirmed=False, status='open', severity='critical')

        vulns += vulnerability_factory.create_batch(3, workspace=self.first_object,
                                                    confirmed=True, status='open', severity='high')

        vulns += vulnerability_web_factory.create_batch(2, workspace=second_workspace,
                                                    confirmed=True, status='open', severity='medium')

        vulns += vulnerability_web_factory.create_batch(2, workspace=second_workspace,
                                                    confirmed=True, status='open', severity='low')

        session.add_all(vulns)
        session.commit()
        res = test_client.get('/v3/ws?histogram=true')
        assert res.status_code == 200
        firs_ws = [ws['histogram'] for ws in res.json if ws['name'] == self.first_object.name]
        assert len(firs_ws[0]) == 20
        ws_histogram = firs_ws[0]
        for ws_date in ws_histogram:
            if ws_date['date'] == date.today().strftime("%Y-%m-%d"):
                assert ws_date['medium'] == 0
                assert ws_date['high'] == 3
                assert ws_date['critical'] == 8
                assert ws_date['confirmed'] == 3
            else:
                assert ws_date['medium'] == 0
                assert ws_date['high'] == 0
                assert ws_date['critical'] == 0
                assert ws_date['confirmed'] == 0

        second_ws = [ws['histogram'] for ws in res.json if ws['name'] == second_workspace.name]
        assert len(second_ws[0]) == 20
        ws_histogram = second_ws[0]
        for ws_date in ws_histogram:
            if ws_date['date'] == date.today().strftime("%Y-%m-%d"):
                assert ws_date['medium'] == 2
                assert ws_date['high'] == 0
                assert ws_date['critical'] == 0
                assert ws_date['confirmed'] == 2
            else:
                assert ws_date['medium'] == 0
                assert ws_date['high'] == 0
                assert ws_date['critical'] == 0
                assert ws_date['confirmed'] == 0

        res = test_client.get('/v3/ws?histogram=True&histogram_days=a')
        assert res.status_code == 200
        firs_ws = [ws['histogram'] for ws in res.json if ws['name'] == self.first_object.name]
        assert len(firs_ws[0]) == 20

        res = test_client.get('/v3/ws?histogram=true&histogram_days=[asdf, "adsf"]')
        assert res.status_code == 200
        firs_ws = [ws['histogram'] for ws in res.json if ws['name'] == self.first_object.name]
        assert len(firs_ws[0]) == 20

        res = test_client.get('/v3/ws?histogram=true&histogram_days=[asdf, "adsf"]')
        assert res.status_code == 200
        firs_ws = [ws['histogram'] for ws in res.json if ws['name'] == self.first_object.name]
        assert len(firs_ws[0]) == 20

        res = test_client.get('/v3/ws?histogram=true&histogram_days=5')
        assert res.status_code == 200
        firs_ws = [ws['histogram'] for ws in res.json if ws['name'] == self.first_object.name]
        assert len(firs_ws[0]) == 5

        res = test_client.get('/v3/ws?histogram=true&histogram_days=365')
        assert res.status_code == 200
        firs_ws = [ws['histogram'] for ws in res.json if ws['name'] == self.first_object.name]
        assert len(firs_ws[0]) == 365

        res = test_client.get('/v3/ws?histogram=asdf&histogram_days=365')
        assert res.status_code == 200
        for ws in res.json:
            assert 'histogram' not in ws

    @pytest.mark.parametrize('querystring', [
        '?status=closed'
    ])
    def test_vuln_count_closed(self,
                        vulnerability_factory,
                        vulnerability_web_factory,
                        test_client,
                        session,
                        querystring):
        vulns = vulnerability_factory.create_batch(8, workspace=self.first_object,
                                                   confirmed=False, status='open', severity='informational')

        vulns += vulnerability_factory.create_batch(3, workspace=self.first_object,
                                                    confirmed=True, status='closed', severity='informational')

        vulns += vulnerability_web_factory.create_batch(2, workspace=self.first_object,
                                                    confirmed=True, status='open', severity='informational')

        session.add_all(vulns)
        session.commit()
        res = test_client.get(urljoin(self.url(self.first_object), querystring))
        assert res.status_code == 200
        assert res.json['stats']['code_vulns'] == 0
        assert res.json['stats']['web_vulns'] == 0
        assert res.json['stats']['std_vulns'] == 3
        assert res.json['stats']['critical_vulns'] == 0
        assert res.json['stats']['info_vulns'] == 3
        assert res.json['stats']['total_vulns'] == 3
        assert res.json['stats']['opened_vulns'] == 0
        assert res.json['stats']['confirmed_vulns'] == 3

    @pytest.mark.parametrize('querystring', [
        '?status=asdfss',
        '?status=close',
        '?status=',
        '?status=\' or 1 == 1',
    ])
    def test_vuln_count_invalid_status(self,
                        vulnerability_factory,
                        vulnerability_web_factory,
                        test_client,
                        session,
                        querystring):
        vulns = vulnerability_factory.create_batch(8, workspace=self.first_object,
                                                   confirmed=False, status='open')

        vulns += vulnerability_factory.create_batch(3, workspace=self.first_object,
                                                    confirmed=True, status='closed')

        vulns += vulnerability_web_factory.create_batch(2, workspace=self.first_object,
                                                    confirmed=True, status='open')

        session.add_all(vulns)
        session.commit()
        res = test_client.get(urljoin(self.url(self.first_object), querystring))
        assert res.status_code == 200
        assert res.json['stats']['code_vulns'] == 0
        assert res.json['stats']['web_vulns'] == 2
        assert res.json['stats']['std_vulns'] == 11
        assert res.json['stats']['total_vulns'] == 13

    @pytest.mark.parametrize('querystring', [
        '?confirmed=1',
        '?confirmed=true'
    ])
    def test_vuln_count_confirmed(self,
                                  vulnerability_factory,
                                  test_client,
                                  session,
                                  querystring):
        vulns = vulnerability_factory.create_batch(8, workspace=self.first_object,
                                                   confirmed=False)
        vulns += vulnerability_factory.create_batch(5, workspace=self.first_object,
                                                    confirmed=True)
        session.add_all(vulns)
        session.commit()
        res = test_client.get(urljoin(self.url(self.first_object), querystring))
        assert res.status_code == 200
        assert res.json['stats']['total_vulns'] == 5

    @pytest.mark.parametrize('querystring', [
        '?confirmed=0',
        '?confirmed=false'
    ])
    def test_vuln_count_confirmed_2(self, vulnerability_factory, test_client, session, querystring):
        vulns = vulnerability_factory.create_batch(8, workspace=self.first_object,
                                                   confirmed=False, severity='critical', status='open')
        vulns += vulnerability_factory.create_batch(5, workspace=self.first_object,
                                                    confirmed=True, status='open')
        session.add_all(vulns)
        session.commit()
        res = test_client.get(self.url(self.first_object) + querystring)
        assert res.status_code == 200
        assert res.json['stats']['std_vulns'] == 8
        assert res.json['stats']['critical_vulns'] == 8
        assert res.json['stats']['info_vulns'] == 0
        assert res.json['stats']['opened_vulns'] == 13
        assert res.json['stats']['confirmed_vulns'] == 0
        assert res.json['stats']['total_vulns'] == 8

    def test_create_fails_with_valid_duration(self, session, test_client):
        workspace_count_previous = session.query(Workspace).count()
        start_date = int(time.time()) * 1000
        end_date = start_date + 86400000
        duration = {'start_date': start_date, 'end_date': end_date}
        raw_data = {'name': 'somethingdarkside', 'duration': duration}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 201
        assert workspace_count_previous + 1 == session.query(Workspace).count()
        assert res.json['duration']['start_date'] == start_date
        assert res.json['duration']['end_date'] == end_date

    def test_create_fails_with_mayus(self, session, test_client):
        workspace_count_previous = session.query(Workspace).count()
        raw_data = {'name': 'sWtr'}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 400
        assert workspace_count_previous == session.query(Workspace).count()

    def test_create_fails_with_special_character(self, session, test_client):
        workspace_count_previous = session.query(Workspace).count()
        raw_data = {'name': '$wtr'}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 400
        assert workspace_count_previous == session.query(Workspace).count()

    def test_create_with_initial_number(self, session, test_client):
        workspace_count_previous = session.query(Workspace).count()
        raw_data = {'name': '2$wtr'}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 201
        assert workspace_count_previous + 1 == session.query(Workspace).count()

    def test_create_fails_with_invalid_duration_start_type(self,
                                                           session,
                                                           test_client):
        workspace_count_previous = session.query(Workspace).count()
        start_date = 'this should clearly fail'
        duration = {'start_date': start_date, 'end_date': 86400000}
        raw_data = {'name': 'somethingdarkside', 'duration': duration}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 400
        assert workspace_count_previous == session.query(Workspace).count()

    @pytest.mark.xfail(reason="Filter not implemented yet")
    def test_create_fails_with_invalid_duration_start_after_end(self,
                                                                session,
                                                                test_client):
        workspace_count_previous = session.query(Workspace).count()
        start_date = int(time.time()) * 1000
        duration = {'start_date': start_date, 'end_date': start_date - 86400000}
        raw_data = {'name': 'somethingdarkside', 'duration': duration}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 400
        assert workspace_count_previous == session.query(Workspace).count()

    def test_create_fails_with_forward_slash(self, session, test_client):
        workspace_count_previous = session.query(Workspace).count()
        raw_data = {'name': 'swtr/'}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 400
        assert workspace_count_previous == session.query(Workspace).count()

    def test_create_with_description(self, session, test_client):
        description = 'darkside'
        raw_data = {'name': 'something', 'description': description}
        workspace_count_previous = session.query(Workspace).count()
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 201
        assert workspace_count_previous + 1 == session.query(Workspace).count()
        assert res.json['description'] == description

    @pytest.mark.parametrize("stat_name", [
        'credentials', 'services', 'web_vulns', 'code_vulns', 'std_vulns',
        'total_vulns'
    ])
    def test_create_stat_is_zero(self, test_client, stat_name):
        raw_data = {'name': 'something', 'description': ''}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 201
        assert res.json['stats'][stat_name] == 0

    def test_update_stats(self, workspace, session, test_client,
                          vulnerability_factory,
                          vulnerability_web_factory):
        vulns = vulnerability_factory.create_batch(10, workspace=workspace)
        vulns += vulnerability_web_factory.create_batch(5, workspace=workspace)
        session.add_all(vulns)
        session.commit()
        raw_data = {'name': 'something', 'description': ''}
        res = test_client.put(self.url(obj=workspace), data=raw_data)
        assert res.status_code == 200
        assert res.json['stats']['web_vulns'] == 5
        assert res.json['stats']['std_vulns'] == 10
        assert res.json['stats']['total_vulns'] == 15

    def test_create_with_scope(self, session, test_client):
        desired_scope = [
            'www.google.com',
            '127.0.0.1'
        ]
        raw_data = {'name': 'something', 'description': 'test',
                    'scope': desired_scope}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 201
        assert set(res.json['scope']) == set(desired_scope)
        workspace = Workspace.query.get(res.json['id'])
        assert {s.name for s in workspace.scope} == set(desired_scope)

    def test_update_with_scope(self, session, test_client, workspace):
        session.add(Scope(name='test.com', workspace=workspace))
        session.add(Scope(name='www.google.com', workspace=workspace))
        desired_scope = [
            'www.google.com',
            '127.0.0.1'
        ]
        raw_data = {'name': 'something', 'description': 'test',
                    'scope': desired_scope}
        res = test_client.put(self.url(obj=workspace), data=raw_data)
        assert res.status_code == 200
        assert set(res.json['scope']) == set(desired_scope)
        assert {s.name for s in workspace.scope} == set(desired_scope)

    @pytest.mark.skip_sql_dialect('sqlite')
    def test_list_retrieves_all_items_from(self, test_client, logged_user):
        super().test_list_retrieves_all_items_from(test_client, logged_user)

    def test_workspace_activation(self, test_client, workspace, session):
        workspace.active = False
        session.add(workspace)
        session.commit()
        res = test_client.patch(self.url(workspace), data={'active': True})
        assert res.status_code == 200

        res = test_client.get(self.url(workspace))
        active = res.json.get('active')
        assert active

        active_query = session.query(Workspace).filter_by(id=workspace.id).first().active
        assert active_query

    def test_workspace_deactivation(self, test_client, workspace, session):
        workspace.active = True
        session.add(workspace)
        session.commit()
        res = test_client.patch(self.url(workspace), data={'active': False})
        assert res.status_code == 200

        res = test_client.get(self.url(workspace))
        active = res.json.get('active')
        assert not active

        active_query = session.query(Workspace).filter_by(id=workspace.id).first().active
        assert not active_query

    def test_create_fails_with_start_date_greater_than_end_date(self,
                                                           session,
                                                           test_client):
        workspace_count_previous = session.query(Workspace).count()
        duration = {'start_date': 1563638577, 'end_date': 1563538577}
        raw_data = {'name': 'somethingdarkside', 'duration': duration}
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 400
        assert workspace_count_previous == session.query(Workspace).count()