Merge pull request #896 from dfir-iris/api_v2_get_user_profile

Api v2 get user profile

Author: whikernel

source/app/blueprints/rest/profile_routes.py

@@ -181,6 +181,7 @@ def profile_refresh_permissions_and_ac():
 
 
 @profile_rest_blueprint.route('/user/whoami', methods=['GET'])
+@endpoint_deprecated('GET', '/api/v2/me')
 @ac_api_requires()
 def profile_whoami():
     """Returns the current user's profile"""

source/app/blueprints/rest/v2/auth.py

@@ -35,7 +35,6 @@
 from app.blueprints.rest.endpoints import response_api_success
 from app.business.auth import validate_ldap_login, validate_local_login, return_authed_user_info, generate_auth_tokens
 from app.iris_engine.utils.tracker import track_activity
-from app.schema.marshables import UserSchema
 
 
 auth_blueprint = Blueprint('auth', __name__, url_prefix='/auth')
@@ -112,23 +111,6 @@ def logout():
     return redirect(not_authenticated_redirection_url('/'))
 
 
-# TODO - We should have /api/v2/users/{identifier}. For now keeping it since the route doesn't exist elsewhere
-@auth_blueprint.route('/whoami', methods=['GET'])
-def whoami():
-    """
-    Returns information about the currently authenticated user.
-    """
-
-    # Ensure we are authenticated
-    if not iris_current_user.is_authenticated:
-        return response_api_error("Unauthenticated")
-
-    # Return the current_user dict
-    return response_api_success(data=UserSchema(only=[
-        'id', 'user_name', 'user_login', 'user_email'
-    ]).dump(iris_current_user))
-
-
 @auth_blueprint.post('/refresh-token')
 def refresh_token_endpoint():
     """

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

@@ -39,114 +39,133 @@
 from app.business.evidences import evidences_update
 from app.business.evidences import evidences_filter
 from app.business.evidences import evidences_delete
-from app.models.models import CaseReceivedFile
 
 
-case_evidences_blueprint = Blueprint('case_evidences_rest_v2', __name__, url_prefix='/<int:case_identifier>/evidences')
+class EvidencesOperations:
 
+    def __init__(self):
+        self._schema = CaseEvidenceSchema()
 
-@case_evidences_blueprint.get('')
-@ac_api_requires()
-def get_evidences(case_identifier):
-    if not cases_exists(case_identifier):
-        return response_api_not_found()
+    @staticmethod
+    def _get_evidence_in_case(identifier, case_identifier):
+        evidence = evidences_get(identifier)
+        if evidence.case_id != case_identifier:
+            raise BusinessProcessingError(f'Evidence {evidence.id} does not belong to case {case_identifier}')
+        return evidence
 
-    if not ac_fast_check_current_user_has_case_access(case_identifier, [CaseAccessLevel.read_only, CaseAccessLevel.full_access]):
-        return ac_api_return_access_denied(caseid=case_identifier)
+    def list(self, case_identifier):
+        if not cases_exists(case_identifier):
+            return response_api_not_found()
 
-    pagination_parameters = parse_pagination_parameters(request, default_order_by='date_added', default_direction='desc')
-    try:
-        evidences = evidences_filter(case_identifier, pagination_parameters)
+        if not ac_fast_check_current_user_has_case_access(case_identifier,
+                                                          [CaseAccessLevel.read_only, CaseAccessLevel.full_access]):
+            return ac_api_return_access_denied(caseid=case_identifier)
 
-        evidence_schema = CaseEvidenceSchema()
-        return response_api_paginated(evidence_schema, evidences)
-    except BusinessProcessingError as e:
-        return response_api_error(e.get_message(), data=e.get_data())
+        pagination_parameters = parse_pagination_parameters(request, default_order_by='date_added',
+                                                            default_direction='desc')
+        try:
+            evidences = evidences_filter(case_identifier, pagination_parameters)
 
+            return response_api_paginated(self._schema, evidences)
+        except BusinessProcessingError as e:
+            return response_api_error(e.get_message(), data=e.get_data())
 
-@case_evidences_blueprint.post('')
-@ac_api_requires()
-def create_evidence(case_identifier):
+    def create(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(caseid=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(caseid=case_identifier)
 
-    try:
-        evidence = evidences_create(case_identifier, request.get_json())
+        try:
+            evidence = evidences_create(case_identifier, request.get_json())
 
-        evidence_schema = CaseEvidenceSchema()
-        return response_api_created(evidence_schema.dump(evidence))
-    except BusinessProcessingError as e:
-        return response_api_error(e.get_message(), data=e.get_data())
+            return response_api_created(self._schema.dump(evidence))
+        except BusinessProcessingError as e:
+            return response_api_error(e.get_message(), data=e.get_data())
 
+    def get(self, case_identifier, identifier):
+        if not cases_exists(case_identifier):
+            return response_api_not_found()
 
-@case_evidences_blueprint.get('/<int:identifier>')
-@ac_api_requires()
-def get_evidence(case_identifier, identifier):
-    if not cases_exists(case_identifier):
-        return response_api_not_found()
+        try:
+            evidence = self._get_evidence_in_case(identifier, case_identifier)
 
-    try:
-        evidence = evidences_get(identifier)
-        _check_evidence_and_case_identifier_match(evidence, case_identifier)
+            if not ac_fast_check_current_user_has_case_access(evidence.case_id,
+                                                              [CaseAccessLevel.read_only, CaseAccessLevel.full_access]):
+                return ac_api_return_access_denied(caseid=evidence.case_id)
 
-        if not ac_fast_check_current_user_has_case_access(evidence.case_id, [CaseAccessLevel.read_only, CaseAccessLevel.full_access]):
-            return ac_api_return_access_denied(caseid=evidence.case_id)
+            return response_api_success(self._schema.dump(evidence))
+        except ObjectNotFoundError:
+            return response_api_not_found()
+        except BusinessProcessingError as e:
+            return response_api_error(e.get_message(), data=e.get_data())
 
-        evidence_schema = CaseEvidenceSchema()
-        return response_api_success(evidence_schema.dump(evidence))
-    except ObjectNotFoundError:
-        return response_api_not_found()
-    except BusinessProcessingError as e:
-        return response_api_error(e.get_message(), data=e.get_data())
+    def update(self, case_identifier, identifier):
+        if not cases_exists(case_identifier):
+            return response_api_not_found()
 
+        try:
+            evidence = self._get_evidence_in_case(identifier, case_identifier)
+            if not ac_fast_check_current_user_has_case_access(evidence.case_id, [CaseAccessLevel.full_access]):
+                return ac_api_return_access_denied(caseid=evidence.case_id)
 
-@case_evidences_blueprint.put('/<int:identifier>')
-@ac_api_requires()
-def update_evidence(case_identifier, identifier):
-    if not cases_exists(case_identifier):
-        return response_api_not_found()
+            evidence = evidences_update(evidence, request.get_json())
 
-    try:
-        evidence = evidences_get(identifier)
-        if not ac_fast_check_current_user_has_case_access(evidence.case_id, [CaseAccessLevel.full_access]):
-            return ac_api_return_access_denied(caseid=evidence.case_id)
-        _check_evidence_and_case_identifier_match(evidence, case_identifier)
+            result = self._schema.dump(evidence)
+            return response_api_success(result)
+        except ObjectNotFoundError:
+            return response_api_not_found()
+        except BusinessProcessingError as e:
+            return response_api_error(e.get_message(), data=e.get_data())
 
-        evidence = evidences_update(evidence, request.get_json())
+    def delete(self, case_identifier, identifier):
+        if not cases_exists(case_identifier):
+            return response_api_not_found()
 
-        schema = CaseEvidenceSchema()
-        result = schema.dump(evidence)
-        return response_api_success(result)
-    except ObjectNotFoundError:
-        return response_api_not_found()
-    except BusinessProcessingError as e:
-        return response_api_error(e.get_message(), data=e.get_data())
+        try:
+            evidence = self._get_evidence_in_case(identifier, case_identifier)
+            if not ac_fast_check_current_user_has_case_access(evidence.case_id, [CaseAccessLevel.full_access]):
+                return ac_api_return_access_denied(caseid=evidence.case_id)
 
+            evidences_delete(evidence)
 
-@case_evidences_blueprint.delete('/<int:identifier>')
+            return response_api_deleted()
+        except ObjectNotFoundError:
+            return response_api_not_found()
+        except BusinessProcessingError as e:
+            return response_api_error(e.get_message(), data=e.get_data())
+
+
+evidences_operations = EvidencesOperations()
+case_evidences_blueprint = Blueprint('case_evidences_rest_v2', __name__, url_prefix='/<int:case_identifier>/evidences')
+
+
+@case_evidences_blueprint.get('')
 @ac_api_requires()
-def delete_evidence(case_identifier, identifier):
-    if not cases_exists(case_identifier):
-        return response_api_not_found()
+def get_evidences(case_identifier):
+    return evidences_operations.list(case_identifier)
+
+
+@case_evidences_blueprint.post('')
+@ac_api_requires()
+def create_evidence(case_identifier):
+    return evidences_operations.create(case_identifier)
 
-    try:
-        evidence = evidences_get(identifier)
-        _check_evidence_and_case_identifier_match(evidence, case_identifier)
-        if not ac_fast_check_current_user_has_case_access(evidence.case_id, [CaseAccessLevel.full_access]):
-            return ac_api_return_access_denied(caseid=evidence.case_id)
 
-        evidences_delete(evidence)
+@case_evidences_blueprint.get('/<int:identifier>')
+@ac_api_requires()
+def get_evidence(case_identifier, identifier):
+    return evidences_operations.get(case_identifier, identifier)
 
-        return response_api_deleted()
-    except ObjectNotFoundError:
-        return response_api_not_found()
-    except BusinessProcessingError as e:
-        return response_api_error(e.get_message(), data=e.get_data())
+
+@case_evidences_blueprint.put('/<int:identifier>')
+@ac_api_requires()
+def update_evidence(case_identifier, identifier):
+    return evidences_operations.update(case_identifier, identifier)
 
 
-def _check_evidence_and_case_identifier_match(evidence: CaseReceivedFile, case_identifier):
-    if evidence.case_id != case_identifier:
-        raise BusinessProcessingError(f'Evidence {evidence.id} does not belong to case {case_identifier}')
+@case_evidences_blueprint.delete('/<int:identifier>')
+@ac_api_requires()
+def delete_evidence(case_identifier, identifier):
+    return evidences_operations.delete(case_identifier, identifier)

source/app/blueprints/rest/v2/profile.py

@@ -35,6 +35,11 @@ def __init__(self):
         self._schema = UserSchemaForAPIV2()
         self._update_request_schema = UserSchemaForAPIV2(exclude=['user_is_service_account', 'user_active', 'uuid'])
 
+    def get(self):
+        user = users_get(iris_current_user.id)
+        result = self._schema.dump(user)
+        return response_api_success(result)
+
     def update(self):
         try:
             user = users_get(iris_current_user.id)
@@ -52,6 +57,12 @@ def update(self):
 profile_blueprint = Blueprint('profile_rest_v2', __name__, url_prefix='/me')
 
 
+@profile_blueprint.get('')
+@ac_api_requires()
+def get_profile():
+    return profile_operations.get()
+
+
 @profile_blueprint.put('')
 @ac_api_requires()
 def update_profile():

tests/tests_rest_profile.py

@@ -28,6 +28,10 @@ def setUp(self) -> None:
     def tearDown(self):
         self._subject.clear_database()
 
+    def test_get_me_should_return_200(self):
+        response = self._subject.get('/api/v2/me')
+        self.assertEqual(200, response.status_code)
+
     def test_update_me_should_return_200(self):
         response = self._subject.update('/api/v2/me', {})
         self.assertEqual(200, response.status_code)