Merge pull request #931 from dfir-iris/v2_api_notes_directories_filter

Api v2 notes directories filter

Author: c8y3

source/app/blueprints/rest/case/case_notes_routes.py

@@ -366,6 +366,7 @@ def case_get_notes_group(cur_id, caseid):
 
 
 @case_notes_rest_blueprint.route('/case/notes/directories/filter', methods=['GET'])
+@endpoint_deprecated('GET', '/api/v2/cases/{case_identifier}/notes-directories')
 @ac_requires_case_identifier(CaseAccessLevel.read_only, CaseAccessLevel.full_access)
 @ac_api_requires()
 def case_filter_notes_directories(caseid):

source/app/blueprints/rest/v2/case_routes/evidences.py

@@ -20,7 +20,8 @@
 from flask import request
 from marshmallow import ValidationError
 
-from app.blueprints.access_controls import ac_api_requires, ac_fast_check_current_user_has_case_access
+from app.blueprints.access_controls import ac_api_requires
+from app.blueprints.access_controls import ac_fast_check_current_user_has_case_access
 from app.models.authorization import CaseAccessLevel
 from app.business.errors import BusinessProcessingError
 from app.business.errors import ObjectNotFoundError

source/app/blueprints/rest/v2/case_routes/notes_directories.py

@@ -20,28 +20,34 @@
 from flask import request
 from marshmallow import ValidationError
 
-from app.blueprints.access_controls import ac_api_requires, ac_fast_check_current_user_has_case_access
+from app.blueprints.access_controls import ac_api_requires
+from app.blueprints.access_controls import ac_fast_check_current_user_has_case_access
+from app.blueprints.rest.parsing import parse_pagination_parameters
 from app.blueprints.rest.endpoints import response_api_created
 from app.blueprints.rest.endpoints import response_api_success
 from app.blueprints.rest.endpoints import response_api_deleted
 from app.blueprints.rest.endpoints import response_api_error
 from app.blueprints.rest.endpoints import response_api_not_found
+from app.blueprints.rest.endpoints import response_api_paginated
 from app.blueprints.access_controls import ac_api_return_access_denied
 from app.business.errors import ObjectNotFoundError
 from app.business.errors import BusinessProcessingError
 from app.schema.marshables import CaseNoteDirectorySchema
+from app.schema.marshables import SearchCaseNoteDirectorySchema
 from app.business.notes_directories import notes_directories_create
 from app.business.notes_directories import notes_directories_get
 from app.business.notes_directories import notes_directories_update
 from app.business.notes_directories import notes_directories_delete
 from app.business.cases import cases_exists
+from app.business.notes_directories import notes_directories_filter
 from app.models.authorization import CaseAccessLevel
 
 
 class NotesDirectories:
 
     def __init__(self):
         self._schema = CaseNoteDirectorySchema()
+        self._schema_search = SearchCaseNoteDirectorySchema(exclude=('parent_id', 'case_id'))
 
     @staticmethod
     def _get_note_directory_in_case(identifier, case_identifier):
@@ -53,6 +59,16 @@ def _get_note_directory_in_case(identifier, case_identifier):
     def _load(self, request_data, **kwargs):
         return self._schema.load(request_data, **kwargs)
 
+    def search(self, case_identifier):
+        if not cases_exists(case_identifier):
+            return response_api_not_found()
+        if not ac_fast_check_current_user_has_case_access(case_identifier, [CaseAccessLevel.full_access]):
+            return ac_api_return_access_denied(case_identifier)
+
+        pagination_parameters = parse_pagination_parameters(request, default_order_by='name', default_direction='asc')
+        directories = notes_directories_filter(case_identifier, pagination_parameters)
+        return response_api_paginated(self._schema_search, directories)
+
     def create(self, case_identifier):
         if not cases_exists(case_identifier):
             return response_api_not_found()
@@ -69,9 +85,8 @@ def create(self, case_identifier):
             directory = self._load(request_data)
 
             notes_directories_create(directory)
-            result = self._schema.dump(directory)
 
-            return response_api_created(result)
+            return response_api_created(self._schema.dump(directory))
         except ValidationError as e:
             return response_api_error('Data error', data=e.normalized_messages())
 
@@ -85,8 +100,7 @@ def get(self, case_identifier, identifier):
         try:
             note_directory = self._get_note_directory_in_case(identifier, case_identifier)
 
-            result = self._schema.dump(note_directory)
-            return response_api_success(result)
+            return response_api_success(self._schema.dump(note_directory))
         except ObjectNotFoundError:
             return response_api_not_found()
         except BusinessProcessingError as e:
@@ -107,8 +121,7 @@ def update(self, case_identifier, identifier):
                 self._schema.verify_parent_id(request_data['parent_id'], case_id=case_identifier, current_id=identifier)
             new_directory = self._load(request_data, instance=directory, partial=True)
             notes_directories_update(new_directory)
-            result = self._schema.dump(new_directory)
-            return response_api_success(result)
+            return response_api_success(self._schema.dump(new_directory))
         except ValidationError as e:
             return response_api_error('Data error', data=e.normalized_messages())
         except ObjectNotFoundError:
@@ -154,3 +167,9 @@ def update_note_directory(case_identifier, identifier):
 @ac_api_requires()
 def delete_note_directory(case_identifier, identifier):
     return notes_directories.delete(case_identifier, identifier)
+
+
+@case_notes_directories_blueprint.get('')
+@ac_api_requires()
+def get_note_directory_filter(case_identifier):
+    return notes_directories.search(case_identifier)

source/app/business/notes_directories.py

@@ -22,6 +22,12 @@
 from app.business.errors import ObjectNotFoundError
 from app.datamgmt.case.case_notes_db import get_directory
 from app.datamgmt.case.case_notes_db import delete_directory
+from app.datamgmt.case.case_notes_db import paginate_notes_directories
+from app.models.pagination_parameters import PaginationParameters
+
+
+def notes_directories_filter(case_identifier: int, pagination_parameters: PaginationParameters):
+    return paginate_notes_directories(case_identifier, pagination_parameters)
 
 
 def notes_directories_create(directory: NoteDirectory):

source/app/datamgmt/case/case_notes_db.py

@@ -20,13 +20,15 @@
 from sqlalchemy import or_
 from sqlalchemy.exc import IntegrityError
 from datetime import datetime
+from flask_sqlalchemy.pagination import Pagination
 
 from app import db
 from app.datamgmt.persistence_error import PersistenceError
 from app.blueprints.iris_user import iris_current_user
 from app.datamgmt.manage.manage_attribute_db import get_default_custom_attributes
 from app.datamgmt.states import update_notes_state
-from app.models.comments import Comments, NotesComments
+from app.models.comments import Comments
+from app.models.comments import NotesComments
 from app.models.models import NoteDirectory
 from app.models.models import NoteRevisions
 from app.models.models import Notes
@@ -35,6 +37,8 @@
 from app.models.authorization import User
 from app.models.cases import Cases
 from app.models.models import Client
+from app.models.pagination_parameters import PaginationParameters
+from app.datamgmt.filtering import paginate
 
 
 def get_note(note_id):
@@ -164,7 +168,7 @@ def update_note_revision(note: Notes) -> bool:
         raise PersistenceError(e)
 
 
-def add_note(note_title, creation_date, user_id, caseid, directory_id, note_content=""):
+def add_note(note_title, creation_date, user_id, caseid, directory_id, note_content=''):
     note = Notes()
     note.note_title = note_title
     note.note_creationdate = note.note_lastupdate = creation_date
@@ -279,7 +283,7 @@ def add_note_group(group_title, caseid, userid, creationdate):
     db.session.commit()
 
     if group_title == '':
-        ng.group_title = "New notes group"
+        ng.group_title = 'New notes group'
 
     db.session.commit()
 
@@ -420,7 +424,7 @@ def delete_note_comment(note_id, comment_id):
         Comments.comment_user_id == iris_current_user.id
     ).first()
     if not comment:
-        return False, "You are not allowed to delete this comment"
+        return False, 'You are not allowed to delete this comment'
 
     NotesComments.query.filter(
         NotesComments.comment_note_id == note_id,
@@ -430,7 +434,7 @@ def delete_note_comment(note_id, comment_id):
     db.session.delete(comment)
     db.session.commit()
 
-    return True, "Comment deleted"
+    return True, 'Comment deleted'
 
 
 def get_directories_with_note_count(case_id):
@@ -454,6 +458,12 @@ def get_directories_with_note_count(case_id):
     return directories_with_note_count
 
 
+def paginate_notes_directories(case_id, pagination_parameters: PaginationParameters) -> Pagination:
+    query = NoteDirectory.query.filter_by(case_id=case_id)
+
+    return paginate(NoteDirectory, pagination_parameters, query)
+
+
 def get_directory_with_note_count(directory):
     note_count = Notes.query.filter_by(directory_id=directory.id).count()
 

source/app/datamgmt/case/case_rfiles_db.py

@@ -25,10 +25,11 @@
 from app.datamgmt.manage.manage_attribute_db import get_default_custom_attributes
 from app.datamgmt.states import update_evidences_state
 from app.models.models import CaseReceivedFile
-from app.models.comments import Comments, EvidencesComments
+from app.models.comments import Comments
+from app.models.comments import EvidencesComments
 from app.models.authorization import User
 from app.models.pagination_parameters import PaginationParameters
-from app.datamgmt.conversions import convert_sort_direction
+from app.datamgmt.filtering import paginate
 
 
 def get_rfiles(caseid):
@@ -45,18 +46,8 @@ def get_paginated_evidences(case_identifier, pagination_parameters: PaginationPa
     query = CaseReceivedFile.query.filter(
         CaseReceivedFile.case_id == case_identifier
     )
-    order_func = convert_sort_direction(pagination_parameters.get_direction())
 
-    order_by = pagination_parameters.get_order_by()
-    column = getattr(CaseReceivedFile, order_by)
-
-    query = query.order_by(order_func(column))
-
-    return query.paginate(
-        page=pagination_parameters.get_page(),
-        per_page=pagination_parameters.get_per_page(),
-        error_out=False
-    )
+    return paginate(CaseReceivedFile, pagination_parameters, query)
 
 
 def add_rfile(evidence: CaseReceivedFile, caseid, user_id):

source/app/datamgmt/filtering.py

@@ -209,11 +209,15 @@ def get_filtered_data(model,
             log.exception(e)
             raise BusinessProcessingError(f'Error parsing custom_conditions: {e}')
 
-    # Apply ordering if requested.
+    return paginate(model, pagination_parameters, query)
+
+
+def paginate(model, pagination_parameters: PaginationParameters, query):
     order_by = pagination_parameters.get_order_by()
     if order_by is not None and hasattr(model, order_by):
         order_func = convert_sort_direction(pagination_parameters.get_direction())
-        query = query.order_by(order_func(getattr(model, order_by)))
+        column = getattr(model, order_by)
+        query = query.order_by(order_func(column))
 
     # Paginate and return the results.
     result = query.paginate(

source/app/schema/marshables.py

@@ -231,6 +231,20 @@ def verify_directory_name(self, data: Dict[str, Any], **kwargs: Any) -> Dict[str
         return data
 
 
+class SearchCaseNoteDirectorySchema(CaseNoteDirectorySchema):
+    """Schema for serializing and deserializing SearchCaseNoteDirectory objects.
+
+    This schema defines the fields to include when serializing and deserializing CaseNoteDirectory objects.
+    It includes fields for the CSRF token, directory name, directory description, and directory ID.
+    It also includes a method for verifying the directory name.
+
+    """
+
+    note_count: int = fields.Integer(required=False)
+    subdirectories: List[str] = fields.List(fields.String, required=False)
+    notes:  List[str] = fields.List(fields.String, required=False)
+
+
 class UserSchema(ma.SQLAlchemyAutoSchema):
     """Schema for serializing and deserializing User objects.
 

tests/tests_rest_notes_directories.py

@@ -252,3 +252,37 @@ def test_delete_note_directory_should_return_403_when_user_has_no_access_to_case
         user = self._subject.create_dummy_user()
         response = user.delete(f'/api/v2/cases/{case_identifier}/notes-directories/{identifier}')
         self.assertEqual(403, response.status_code)
+
+    def test_get_notes_directories_filter_should_return_200(self):
+        case_identifier = self._subject.create_dummy_case()
+        body = {'name': 'directory_name'}
+        self._subject.create(f'/api/v2/cases/{case_identifier}/notes-directories', body).json()
+
+        response = self._subject.get(f'/api/v2/cases/{case_identifier}/notes-directories')
+        self.assertEqual(200, response.status_code)
+
+    def test_get_notes_directories_should_return_403_when_user_has_no_access_to_case(self):
+        case_identifier = self._subject.create_dummy_case()
+        body = {'name': 'directory_name'}
+        self._subject.create(f'/api/v2/cases/{case_identifier}/notes-directories', body).json()
+
+        user = self._subject.create_dummy_user()
+        response = user.get(f'/api/v2/cases/{case_identifier}/notes-directories')
+        self.assertEqual(403, response.status_code)
+
+    def test_get_notes_directories_filter_should_return_total(self):
+        case_identifier = self._subject.create_dummy_case()
+        body = {'name': 'directory_name'}
+        self._subject.create(f'/api/v2/cases/{case_identifier}/notes-directories', body).json()
+
+        response = self._subject.get(f'/api/v2/cases/{case_identifier}/notes-directories').json()
+        self.assertEqual(1, response['total'])
+
+    def test_get_notes_directories_filter_should_accept_per_page_query_parameter(self):
+        case_identifier = self._subject.create_dummy_case()
+        body = {'name': 'directory_name'}
+        self._subject.create(f'/api/v2/cases/{case_identifier}/notes-directories', body).json()
+        self._subject.create(f'/api/v2/cases/{case_identifier}/notes-directories', body).json()
+
+        response = self._subject.get(f'/api/v2/cases/{case_identifier}/notes-directories', {'per_page': 1}).json()
+        self.assertEqual(1, len(response['data']))