Add another Python path param case, raise exception for missing input…

… file. (#36)

Signed-off-by: Caroline Russell <[email protected]>

atom_tools/__init__.py

@@ -1,4 +1,4 @@
 """
 A cli, classes and functions for converting an atom slice to a different format
 """
-__version__ = '0.4.2'
+__version__ = '0.4.3'

atom_tools/lib/converter.py

@@ -324,12 +324,11 @@ def _process_methods_helper(self, pattern: str) -> Dict[str, Any]:
         """
         dict_resolved_pattern = jmespath.compile(pattern)
         result = []
-        if dict_resolved_pattern:
+        if matches := dict_resolved_pattern.search(self.usages.content):
             result = [
-                i for i in dict_resolved_pattern.search(self.usages.content)
+                i for i in matches
                 if i.get('resolved_methods')
             ]
-
         resolved: Dict = {}
         for r in result:
             file_name = r['file_name']
@@ -429,7 +428,7 @@ def _paths_object_helper(
         tmp_params: List = []
         py_special_case = False
         orig_ep = ep
-        if ':' in ep:
+        if ':' in ep or '<' in ep:
             ep, py_special_case, tmp_params = self._extract_params(ep)
         if '{' in ep and not py_special_case:
             tmp_params = self.generic_params_helper(ep, orig_ep)

atom_tools/lib/regex_utils.py

@@ -21,6 +21,8 @@ class OpenAPIRegexCollection:
     processed_param = re.compile(r'{(?P<pname>[^\s}]+)}')
     # This regex is used to extract named python parameters that include a type
     py_param = re.compile(r'<(?P<ptype>\w+):(?P<pname>\w+)>')
+    # Secondary python parameter regex
+    py_param_2 = re.compile(r'<(?P<pname>[^\s>]+)>')
     # This regex is used to detect when a path contains a regex set
     detect_regex = re.compile(r'[|$^]|\\[sbigmd]|\\[dhsvwpbagznfrtu\d]|\?P?[<:!]|\{\d|\[\S+]|\Wr\"')
     # This regex is used to extract regexes so we can escape forward slashes not part of the path.
@@ -68,13 +70,18 @@ def py_helper(endpoint: str, regex: OpenAPIRegexCollection) -> Tuple[str, List[D
     """
     params = []
 
-    if matches := regex.py_param.findall(endpoint):
+    if ':' in endpoint and (matches := regex.py_param.findall(endpoint)):
         endpoint = re.sub(regex.py_param, path_param_repl, endpoint)
         for m in matches:
             p = {'in': 'path', 'name': m[1], 'required': True}
             if PY_TYPE_MAPPING.get(m[0]):
                 p['schema'] = {'type': PY_TYPE_MAPPING[m[0]]}
             params.append(p)
+    elif matches := regex.py_param_2.findall(endpoint):
+        endpoint = re.sub(regex.py_param_2, path_param_repl, endpoint)
+        for m in matches:
+            p = {'in': 'path', 'name': m, 'required': True}
+            params.append(p)
     return endpoint, params
 
 

atom_tools/lib/slices.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import sys
 from pathlib import Path
 from typing import Tuple, Dict
 
@@ -65,6 +66,7 @@ def import_slice(filename: str | Path) -> Tuple[Dict, str]:
                 f' json file.'
             )
         except FileNotFoundError:
-            logger.warning(f'Failed to locate the following slice file: {filename}')
+            logger.exception(f'Failed to locate the following slice file: {filename}')
+            sys.exit(1)
         logger.warning('Slice type not recognized.')
         return {}, 'unknown'

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "atom-tools"
-version = "0.4.2"
+version = "0.4.3"
 description = "Collection of tools for use with AppThreat/atom."
 authors = [
   { name = "Caroline Russell", email = "[email protected]" },

test/test_converter.py

@@ -28,7 +28,7 @@ def js_usages_2():
 
 @pytest.fixture
 def py_usages_1():
-    return OpenAPI('openapi3.0.1', 'python', 'test/data/py-airflow-usages.json')
+    return OpenAPI('openapi3.0.1', 'python', 'test/data/py-django-goat-usages.json')
 
 
 @pytest.fixture
@@ -686,63 +686,33 @@ def test_convert_usages(java_usages_1, java_usages_2, js_usages_1, js_usages_2,
                                 'target': {'src\\main\\java\\org\\joychou\\controller\\XStreamRce.java': 23}}}}
     assert len(js_usages_1.convert_usages()) == 134
     assert len(js_usages_2.convert_usages()) == 21
-    assert py_usages_2.convert_usages() == {'/': {},
- '/127.0.0.1': {},
- '/auth/google': {},
- '/google_oauth': {},
- '/logout': {}}
-
-
-# Airflow slice is too large to upload.
-@pytest.mark.skipif(not os.path.exists('test/data/py-airflow-usages.json'), reason="requires test/data/py-airflow-usages.json")
-def test_convert_usages2(py_usages_1):
-    assert py_usages_1.convert_usages() == {
-        '/': {}, '/dags/{dag_id}': {
-        'parameters': [{'in': 'path', 'name': 'dag_id', 'required': True,
-                        'schema': {'type': 'string'}}]},
-        '/dags/{dag_id}/code': {'parameters': [
-            {'in': 'path', 'name': 'dag_id', 'required': True,
-             'schema': {'type': 'string'}}]}, '/dags/{dag_id}/dag_runs': {
-            'parameters': [{'in': 'path', 'name': 'dag_id', 'required': True,
-                            'schema': {'type': 'string'}}]},
-        '/dags/{dag_id}/dag_runs/{'
-        'execution_date}': {'parameters': [
-            {'in': 'path', 'name': 'dag_id', 'required': True,
-             'schema': {'type': 'string'}},
-            {'in': 'path', 'name': 'execution_date', 'required': True,
-             'schema': {'type': 'string'}}]}, '/dags/{dag_id}/dag_runs/{'
-                                              'execution_date}/tasks/{task_id}': {
-            'parameters': [{'in': 'path', 'name': 'dag_id', 'required': True,
-                            'schema': {'type': 'string'}},
-                           {'in': 'path', 'name': 'execution_date',
-                            'required': True, 'schema': {'type': 'string'}},
-                           {'in': 'path', 'name': 'task_id', 'required': True,
-                            'schema': {'type': 'string'}}]},
-        '/dags/{dag_id}/paused': {'parameters': [
-            {'in': 'path', 'name': 'dag_id', 'required': True,
-             'schema': {'type': 'string'}}]},
-        '/dags/{dag_id}/paused/{paused}': {'parameters': [
-            {'in': 'path', 'name': 'dag_id', 'required': True,
-             'schema': {'type': 'string'}},
-            {'in': 'path', 'name': 'paused', 'required': True,
-             'schema': {'type': 'string'}}]},
-        '/dags/{dag_id}/tasks/{task_id}': {'parameters': [
-            {'in': 'path', 'name': 'dag_id', 'required': True,
-             'schema': {'type': 'string'}},
-            {'in': 'path', 'name': 'task_id', 'required': True,
-             'schema': {'type': 'string'}}]}, '/info': {}, '/latest_runs': {},
-        '/lineage/{dag_id}/{'
-        'execution_date}': {'parameters': [
-            {'in': 'path', 'name': 'dag_id', 'required': True,
-             'schema': {'type': 'string'}},
-            {'in': 'path', 'name': 'execution_date', 'required': True,
-             'schema': {'type': 'string'}}]}, '/log/{filename}': {
-            'parameters': [{'in': 'path', 'name': 'filename', 'required': True,
-                            'schema': {'type': 'string'}}]}, '/pools': {},
-        '/pools/{name}': {'parameters': [
-            {'in': 'path', 'name': 'name', 'required': True,
-             'schema': {'type': 'string'}}]}, '/test': {}
-                                            }
+    assert py_usages_1.convert_usages() == {'/admin': {},
+ '/conversation': {},
+ '/conversation/{friend_pk}': {'parameters': [{'in': 'path',
+                                               'name': 'friend_pk',
+                                               'required': True}]},
+ '/dash': {},
+ '/instructions': {},
+ '/landing': {},
+ '/login': {},
+ '/logout': {},
+ '/media/{path}': {'parameters': [{'in': 'path',
+                                   'name': 'path',
+                                   'required': True,
+                                   'schema': {'pattern': '.*',
+                                              'type': 'string'}}]},
+ '/note': {},
+ '/note/{pk}': {'parameters': [{'in': 'path', 'name': 'pk', 'required': True}]},
+ '/profile': {},
+ '/profile-update': {},
+ '/profile/{pk}': {'parameters': [{'in': 'path',
+                                   'name': 'pk',
+                                   'required': True}]},
+ '/sign-up': {},
+ '/vulnerabilities': {},
+ '/write-note': {}}
+    assert py_usages_2.convert_usages() == {'/': {}, '/127.0.0.1': {}, '/auth/google': {},
+                                            '/google_oauth': {}, '/logout': {}}
 
 
 def test_endpoints_to_openapi(java_usages_1):

test/test_slices.py

@@ -44,8 +44,6 @@ def test_usages_class(
 
 
 def test_import_slices(js_usages_1):
-    # Test nonexistent file
-    assert AtomSlice('test/data/js-tornado-usages.json', 'js').content == {}
 
     # Test invalid JSON file
     assert AtomSlice('test/data/invalid.json', 'js').content == {}