detects whether an API request is a transaction (#5)

* detect whether request is an API transaction

* avoid installing as a Python egg

* detect if request is for metadata readonly

* add tests

* update tests

* update tests

* add hooks for metadata endpoints

Dockerfile

@@ -29,7 +29,7 @@ COPY . /app
 # install wis2box_auth
 RUN cd /app \
     && pip3 install -r requirements.txt \
-    && python3 setup.py install
+    && pip3 install -e .
 
 COPY ./entrypoint.sh /entrypoint.sh
 

tests/integration/test_auth.py

@@ -29,9 +29,11 @@
 TOPIC = 'admin'
 TOPIC1 = 'oapi'
 TOPIC2 = 'ui'
+TOPIC3 = 'collections/stations'
 TOKEN = 'test_token'
 TOKEN1 = 'token_1'
 TOKEN2 = '2_test_token'
+TOKEN3 = '3_test_token'
 
 
 def test_no_auth():
@@ -52,6 +54,13 @@ def test_no_auth():
     r = requests.get(URL + '/authorize', headers=headers)
     assert r.status_code == 200
 
+    headers = {
+        'X-Original-URI': f'/oapi/collections/stations/items?token={TOKEN}',
+        'X-Api-Http-Method': 'GET'
+    }
+    r = requests.get(URL + '/authorize', headers=headers)
+    assert r.status_code == 200
+
 
 def test_add_auth():
     '''Test adding wis2box authentication'''
@@ -68,6 +77,10 @@ def test_add_auth():
     r = requests.post(URL + '/add_token', data=data)
     assert r.status_code == 200
 
+    data = {'topic': TOPIC3, 'token': TOKEN3}
+    r = requests.post(URL + '/add_token', data=data)
+    assert r.status_code == 200
+
 
 def test_header_auth():
     '''Test wis2box header authentication'''
@@ -102,6 +115,28 @@ def test_header_auth():
     r = requests.get(URL + '/authorize', headers=headers)
     assert r.status_code == 200
 
+    headers = {
+        'X-Original-URI': f'/{TOPIC3}',
+    }
+    r = requests.get(URL + '/authorize', headers=headers)
+    assert r.status_code == 200
+
+    headers = {
+        'X-Original-URI': f'/{TOPIC3}',
+        'Authorization': f'Bearer {TOKEN3}',
+        'X-Api-Http-Method': 'POST'
+    }
+    r = requests.get(URL + '/authorize', headers=headers)
+    assert r.status_code == 200
+
+    headers = {
+        'X-Original-URI': f'/{TOPIC3}',
+        'Authorization': f'Bearer {TOKEN1}',
+        'X-Api-Http-Method': 'POST'
+    }
+    r = requests.get(URL + '/authorize', headers=headers)
+    assert r.status_code == 401
+
 
 def test_token_auth():
     '''Test wis2box token authentication'''

wis2box_auth/__init__.py

@@ -116,6 +116,9 @@ def extract_topic(topic: str = None) -> bool:
     if any([x in topic for x in ['processes', 'execution']]):
         LOGGER.debug('topic is an API process execution')
         sanitized_topic = topic
+    elif any([x in topic for x in ['collections/stations', 'collections/discovery-metadata']]):  # noqa
+        LOGGER.debug('topic is an API metadata transaction')
+        sanitized_topic = topic
     else:
         sanitized_topic = topic.replace('/', '.')
 

wis2box_auth/app.py

@@ -61,6 +61,18 @@ def authorize():
     request_uri = request.headers.get('X-Original-URI')
     request_ = request.from_values(request_uri)
 
+    metadata_collections = [
+        'discovery-metadata',
+        'stations'
+    ]
+
+    if (request.headers.get('X-Api-Http-Method', 'GET') == 'GET' and
+            any([x in request_uri for x in metadata_collections])):
+        LOGGER.debug('API metadata request')
+        msg = 'Resource is open'
+        LOGGER.debug(msg)
+        return get_response(200, msg)
+
     LOGGER.debug('Extracting topic from request URI')
     resource = extract_topic(request_uri)