Codebase list python-faraday / bd674824-7810-4dbf-8a8b-4c9505581ed5/main tests / test_api_vulnerability_template.py
bd674824-7810-4dbf-8a8b-4c9505581ed5/main

Tree @bd674824-7810-4dbf-8a8b-4c9505581ed5/main (Download .tar.gz)

test_api_vulnerability_template.py @bd674824-7810-4dbf-8a8b-4c9505581ed5/mainraw · history · blame

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

'''

import os
from io import BytesIO
import pytest

from faraday.server.api.modules.vulnerability_template import VulnerabilityTemplateView
from tests import factories
from tests.test_api_non_workspaced_base import (
    ReadOnlyAPITests
)
from faraday.server.models import (
    VulnerabilityTemplate,
    ReferenceTemplate)
from tests.factories import (
    VulnerabilityTemplateFactory,
    ReferenceTemplateFactory,
    CustomFieldsSchemaFactory
)

CURRENT_PATH = os.path.dirname(os.path.abspath(__file__))


@pytest.mark.usefixtures('logged_user')
class TestListVulnerabilityTemplateView(ReadOnlyAPITests):
    model = VulnerabilityTemplate
    factory = factories.VulnerabilityTemplateFactory
    api_endpoint = 'vulnerability_template'
    view_class = VulnerabilityTemplateView

    def test_backwards_json_compatibility(self, test_client, session):
        self.factory.create()
        session.commit()
        res = test_client.get(self.url())
        assert res.status_code == 200
        assert 'rows' in res.json
        for vuln in res.json['rows']:
            assert set([u'id', u'key', u'value', u'doc']) == set(vuln.keys())
            object_properties = [
                u'exploitation',
                u'references',
                u'refs',
                u'name',
                u'cwe',
                u'_rev',
                u'_id',
                u'resolution',
                u'description',
                u'desc'
            ]

            expected = set(object_properties)
            result = set(vuln['doc'].keys())
            assert expected - result == set()

    def _create_post_data_vulnerability_template(self, references):
        data = {
            "exploitation":"high",
            "references":references,
            "name":"name",
            "resolution": "resolution",
            "cwe":"swe",
            "description":"desc"}
        return data

    def test_create_new_vulnerability_template(self, session, test_client):
        vuln_count_previous = session.query(VulnerabilityTemplate).count()
        raw_data = self._create_post_data_vulnerability_template(references='')
        res = test_client.post('/v2/vulnerability_template/', data=raw_data)
        assert res.status_code == 201
        assert isinstance(res.json['_id'], int)
        assert vuln_count_previous + 1 == session.query(VulnerabilityTemplate).count()
        vuln_template = VulnerabilityTemplate.query.get(res.json['_id'])
        assert vuln_template.references == set()

    def test_update_vulnerability_template(self, session, test_client):
        template = self.factory.create()
        session.commit()
        raw_data = self._create_post_data_vulnerability_template(references='')
        res = test_client.put('/v2/vulnerability_template/{0}/'.format(template.id), data=raw_data)
        assert res.status_code == 200
        updated_template = session.query(VulnerabilityTemplate).filter_by(id=template.id).first()
        assert updated_template.name == raw_data['name']
        assert updated_template.severity == raw_data['exploitation']
        assert updated_template.resolution == raw_data['resolution']
        assert updated_template.description == raw_data['description']
        assert updated_template.references == set([])

    @pytest.mark.parametrize('references', [
        ',',
        ',,',
        'a,',
        ['a', 'b', ''],
        {"a": 1},
        {}
    ])
    def test_400_on_invalid_reference(self, session, test_client, references):
        template = self.factory.create()
        session.commit()
        raw_data = self._create_post_data_vulnerability_template(
            references=references)
        res = test_client.put('/v2/vulnerability_template/{0}/'.format(
            template.id), data=raw_data)
        assert res.status_code == 400

    def test_update_vulnerabiliy_template_change_refs(self, session, test_client):
        template = self.factory.create()
        for ref_name in set(['old1', 'old2']):
            ref = ReferenceTemplateFactory.create(name=ref_name)
            self.first_object.reference_template_instances.add(ref)
        session.commit()
        raw_data = self._create_post_data_vulnerability_template(references='new_ref,another_ref')
        res = test_client.put('/v2/vulnerability_template/{0}/'.format(template.id), data=raw_data)
        assert res.status_code == 200
        updated_template = session.query(VulnerabilityTemplate).filter_by(id=template.id).first()
        assert updated_template.name == raw_data['name']
        assert updated_template.severity == raw_data['exploitation']
        assert updated_template.resolution == raw_data['resolution']
        assert updated_template.description == raw_data['description']
        assert updated_template.references == set([u'another_ref', u'new_ref'])

    def test_create_new_vulnerability_template_with_references(self, session, test_client):
        vuln_count_previous = session.query(VulnerabilityTemplate).count()
        raw_data = self._create_post_data_vulnerability_template(references='ref1,ref2')
        res = test_client.post('/v2/vulnerability_template/', data=raw_data)
        assert res.status_code == 201
        assert isinstance(res.json['_id'], int)
        assert set(res.json['refs']) == set(['ref1', 'ref2'])
        assert vuln_count_previous + 1 == session.query(VulnerabilityTemplate).count()
        new_template = session.query(VulnerabilityTemplate).filter_by(id=res.json['_id']).first()
        assert new_template.references == set([u'ref1', u'ref2'])

    def test_delete_vuln_template(self, session, test_client):
        template = self.factory.create()
        vuln_count_previous = session.query(VulnerabilityTemplate).count()
        res = test_client.delete('/v2/vulnerability_template/{0}/'.format(template.id))

        assert res.status_code == 204
        assert vuln_count_previous - 1 == session.query(VulnerabilityTemplate).count()

    def test_create_same_vuln_template_multipe_times_sohuld_raise_409(self, test_client):
        """
            Current test case was found un hackaton. When creating vuln
            template it returned error 500 instead of 409.

        """

        raw_data = {
            "id":123010,
            "cwe": "",
            "description": "test2",
            "desc":"test2",
            "exploitation":"critical",
            "name":"test2",
            "references":[],
            "refs":[],
            "resolution":"",
            "type":"vulnerability_template"
        }
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 201
        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 409

    def test_when_a_template_with_ref_is_deleted_ref_remains_at_database(
            self, session, test_client):
        session.query(ReferenceTemplate).count() == 0
        template = VulnerabilityTemplateFactory.create()
        ref1 = ReferenceTemplateFactory.create()
        template.reference_template_instances.add(ref1)
        session.commit()
        res = test_client.delete(self.url(template))
        assert res.status_code == 204

        assert session.query(ReferenceTemplate).count() == 1

    def test_create_same_vuln_template_with_custom_fields(self, session, test_client):

        custom_field_schema = CustomFieldsSchemaFactory(
            field_name='cvss',
            field_type='str',
            field_display_name='CVSS',
            table_name='vulnerability'
        )
        session.add(custom_field_schema)
        session.commit()

        raw_data = {
            "id":123010,
            "cwe": "",
            "description": "test2",
            "desc":"test2",
            "exploitation":"critical",
            "name":"test2",
            "references":[],
            "refs":[],
            "resolution":"",
            "type": "vulnerability_template",
            "customfields": {
                "cvss": "value",
            }
        }

        res = test_client.post(self.url(), data=raw_data)
        assert res.status_code == 201
        assert res.json['customfields'] == {u'cvss': u'value'}

    def test_update_vuln_template_with_custom_fields(self, session, test_client):

        custom_field_schema = CustomFieldsSchemaFactory(
            field_name='cvss',
            field_type='str',
            field_display_name='CVSS',
            table_name='vulnerability'
        )
        template = VulnerabilityTemplateFactory.create()
        session.add(custom_field_schema)
        session.add(template)
        session.commit()

        raw_data = {
            "cwe": "",
            "description": "test2",
            "desc":"test2",
            "exploitation":"critical",
            "name":"test2",
            "references":[],
            "refs":[],
            "resolution":"",
            "type": "vulnerability_template",
            "customfields": {
                "cvss": "updated value",
            }
        }

        res = test_client.put(self.url(template.id), data=raw_data)
        assert res.status_code == 200
        assert res.json['customfields'] == {u'cvss': u'updated value'}

        vuln_template = session.query(VulnerabilityTemplate).filter_by(id=template.id).first()
        assert vuln_template.custom_fields == {u'cvss': u'updated value'}

    def test_add_vuln_template_from_csv(self, session, test_client, csrf_token):
        expected_created_vuln_template = 1
        file_contents = b"""cwe,name,description,resolution,exploitation,references\n
CWE-119,EN-Improper Restriction of Operations within the Bounds of a Memory Buffer (Type: Class),"The software performs operations on a memory buffer, but it can read from or write to a memory location that is outside of the intended boundary of the buffer.\n
Certain languages allow direct addressing of memory locations and do not automatically ensure that these locations are valid for the memory buffer that is being referenced. This can cause read or write operations to be performed on memory locations that may be associated with other variables, data structures, or internal program data.\n
As a result, an attacker may be able to execute arbitrary code, alter the intended control flow, read sensitive information, or cause the system to crash.",,high,"Writing Secure Code: Chapter 5, ""Public Enemy #1: The Buffer Overrun"" Page 127; Chapter 14, ""Prevent I18N Buffer Overruns"" Page 441\n
Using the Strsafe.h Functions: http://msdn.microsoft.com/en-us/library/ms647466.aspx\n
Safe C String Library v1.0.3: http://www.zork.org/safestr/\n
Address Space Layout Randomization in Windows Vista: http://blogs.msdn.com/michael_howard/archive/2006/05/26/address-space-layout-randomization-in-windows-vista.aspx\n
Limiting buffer overflows with ExecShield: http://www.redhat.com/magazine/009jul05/features/execshield/\n
PaX: http://en.wikipedia.org/wiki/PaX\n
Understanding DEP as a mitigation technology part 1: http://blogs.technet.com/b/srd/archive/2009/06/12/understanding-dep-as-a-mitigation-technology-part-1.aspx\n
The Art of Software Security Assessment: Chapter 5, ""Memory Corruption"", Page 167.\n
The Art of Software Security Assessment: Chapter 5, ""Protection Mechanisms"", Page 189."
"""
        data = {
            'file': (BytesIO(file_contents), 'vulns.csv'),
            'csrf_token': csrf_token
        }
        headers = {'Content-type': 'multipart/form-data'}
        res = test_client.post('/v2/vulnerability_template/bulk_create/',
                               data=data, headers=headers, use_json_data=False)
        assert res.status_code == 200
        assert res.json['vulns_created'] == expected_created_vuln_template

    def test_add_unicode_vuln_template_from_csv(self, session, test_client, csrf_token):
        expected_created_vuln_template = 1
        file_contents = """cwe,name,description,resolution,exploitation,references
        ,ES-Exposición de información a través del listado de directorios,"Estos directorios no deberian estar publicos, pues exponen información sensible del tipo de tecnología utilizada, código de programación, información sobre rutas de acceso a distintos lugares, particularmente en este caso podemos listar toda la información del servidor sin ningun tipo de restricción
        ",Siempre evitar que se puedan listar directorios de manera externa y sin permisos,high,
        """
        data = {
            'file': (BytesIO(file_contents.encode()), 'vulns.csv'),
            'csrf_token': csrf_token
        }
        headers = {'Content-type': 'multipart/form-data'}
        res = test_client.post('/v2/vulnerability_template/bulk_create/',
                               data=data, headers=headers, use_json_data=False)
        assert res.status_code == 200
        assert res.json['vulns_created'] == expected_created_vuln_template

    def test_add_vuln_template_only_required_fields(self, session, test_client, csrf_token):
        expected_created_vuln_template = 1
        file_contents = b"""name,exploitation\n
        "test",high

    """
        data = {
            'file': (BytesIO(file_contents), 'vulns.csv'),
            'csrf_token': csrf_token
        }
        headers = {'Content-type': 'multipart/form-data'}
        res = test_client.post('/v2/vulnerability_template/bulk_create/',
                               data=data, headers=headers, use_json_data=False)
        assert res.status_code == 200
        assert res.json['vulns_created'] == expected_created_vuln_template

    def test_add_vuln_template_missing_required_fields(self, session, test_client, csrf_token):
        expected_created_vuln_template = 1
        file_contents = b"""name,description\n
        "test","description"

    """
        data = {
            'file': (BytesIO(file_contents), 'vulns.csv'),
            'csrf_token': csrf_token
        }
        headers = {'Content-type': 'multipart/form-data'}
        res = test_client.post('/v2/vulnerability_template/bulk_create/',
                               data=data, headers=headers, use_json_data=False)
        assert res.status_code == 400
        assert 'name' not in res.data.decode('utf8')
        assert 'exploitation' in res.data.decode('utf8')

    def test_add_vuln_template_custom_fields_and_some_more_fields(self, session, test_client, csrf_token):

        custom_field_schema = CustomFieldsSchemaFactory(
            field_name='cvss',
            field_type='str',
            field_display_name='CVSS',
            table_name='vulnerability'
        )
        session.add(custom_field_schema)
        session.commit()
        file_contents = b"""name,exploitation,resolution,data,cvss\n
        "test",high,"resolution","technical details","5"
    """
        data = {
            'file': (BytesIO(file_contents), 'vulns.csv'),
            'csrf_token': csrf_token
        }
        headers = {'Content-type': 'multipart/form-data'}
        res = test_client.post('/v2/vulnerability_template/bulk_create/',
                               data=data, headers=headers, use_json_data=False)
        assert res.status_code == 200
        assert res.json['vulns_created'] == 1
        inserted_template = session.query(VulnerabilityTemplate).filter_by(name='test').first()
        assert inserted_template.resolution == 'resolution'
        assert inserted_template.severity == 'high'
        assert inserted_template.data == 'technical details'
        assert 'cvss' in inserted_template.custom_fields
        assert inserted_template.custom_fields['cvss'] == '5'