Merge pull request #51 from automationator/master

Allows creating indicators without the username parameter

services/web/project/api/helpers.py

@@ -1,3 +1,19 @@
+def get_apikey(request):
+    # Get the API key if there is one.
+    # The header should look like:
+    #     Authorization: Apikey blah-blah-blah
+    # So strip off the first 7 characters to get the actual key.
+    authorization = request.headers.get('Authorization')
+    if authorization and 'apikey' in authorization.lower():
+        apikey = authorization[7:]
+
+        # Look up the user by their API key.
+        if apikey:
+            return apikey
+
+    return None
+
+
 def parse_boolean(string, default=False):
     string = str(string).lower()
 

services/web/project/api/routes/indicator.py

@@ -10,7 +10,7 @@
 from project.api import bp
 from project.api.decorators import check_apikey, validate_json, validate_schema
 from project.api.errors import error_response
-from project.api.helpers import parse_boolean
+from project.api.helpers import get_apikey, parse_boolean
 from project.api.schemas import indicator_create, indicator_update
 from project.models import Campaign, Indicator, IndicatorConfidence, IndicatorImpact, IndicatorStatus, IndicatorType, \
     IntelReference, IntelSource, Tag, User
@@ -126,23 +126,35 @@ def create_indicator():
     :status 400: JSON does not match the schema
     :status 401: Invalid role to perform this action
     :status 401: Username is inactive
+    :status 401: You must supply either username or API key
     :status 404: Campaign not found
     :status 404: Confidence not found
     :status 404: Impact not found
     :status 404: Reference not found
     :status 404: Status not found
     :status 404: Tag not found
     :status 404: Type not found
+    :status 404: User not found by API key
     :status 404: Username not found
     :status 409: Indicator already exists
     """
 
     data = request.get_json()
 
-    # Verify the username.
-    user = User.query.filter_by(username=data['username']).first()
-    if not user:
-        return error_response(404, 'User username not found: {}'.format(data['username']))
+    # Verify the user exists.
+    user = None
+    if 'username' in data:
+        user = User.query.filter_by(username=data['username']).first()
+        if not user:
+            return error_response(404, 'User not found by username')
+    else:
+        apikey = get_apikey(request)
+        if apikey:
+            user = User.query.filter_by(apikey=apikey).first()
+            if not user:
+                return error_response(404, 'User not found by API key')
+        else:
+            return error_response(401, 'You must supply either username or API key')
 
     # Verify the user is active.
     if not user.active:

services/web/project/api/schemas/indicator_create.json

@@ -33,6 +33,6 @@
         "username": {"type": "string", "minLength": 1, "maxLength": 255},
         "value": {"type": "string", "minLength": 1, "maxLength": 512}
     },
-    "required": ["username", "type", "value"],
+    "required": ["type", "value"],
     "additionalProperties": false
 }

services/web/project/tests/api/test_indicator.py

@@ -28,13 +28,6 @@ def test_create_schema(client):
     assert request.status_code == 400
     assert response['msg'] == "Request JSON does not match schema: 'type' is a required property"
 
-    # Missing required username parameter
-    data = {'type': 'asdf', 'value': 'asdf'}
-    request = client.post('/api/indicators', json=data)
-    response = json.loads(request.data.decode())
-    assert request.status_code == 400
-    assert response['msg'] == "Request JSON does not match schema: 'username' is a required property"
-
     # Missing required value parameter
     data = {'type': 'asdf', 'username': 'asdf'}
     request = client.post('/api/indicators', json=data)
@@ -329,7 +322,7 @@ def test_create_nonexistent_username(client):
 
     request, response = create_indicator(client, 'asdf', 'asdf', 'this_user_does_not_exist')
     assert request.status_code == 404
-    assert 'User username not found:' in response['msg']
+    assert response['msg'] == 'User not found by username'
 
 
 def test_create_inactive_username(client):
@@ -387,6 +380,66 @@ def test_create_invalid_role(app, client):
     assert response['msg'] == 'Insufficient privileges'
 
 
+def test_create_api_key_instead_of_username(app, client):
+    """ Ensure that supplying an API key instead of username value works """
+
+    app.config['INDICATOR_AUTO_CREATE_INDICATORCONFIDENCE'] = True
+    app.config['INDICATOR_AUTO_CREATE_INDICATORIMPACT'] = True
+    app.config['INDICATOR_AUTO_CREATE_INDICATORSTATUS'] = True
+    app.config['INDICATOR_AUTO_CREATE_INDICATORTYPE'] = True
+
+    # Try a missing API key.
+    data = {'confidence': 'LOW',
+            'impact': 'LOW',
+            'status': 'New',
+            'substring': False,
+            'type': 'Email - Address',
+            'value': 'badguy@evil.com'}
+    request = client.post('/api/indicators', json=data)
+    response = json.loads(request.data.decode())
+    assert request.status_code == 401
+    assert response['msg'] == 'You must supply either username or API key'
+
+    # Try an invalid API key.
+    headers = {'Authorization': 'Apikey ' + 'this-api-key-does-not-exist'}
+    data = {'confidence': 'LOW',
+            'impact': 'LOW',
+            'status': 'New',
+            'substring': False,
+            'type': 'Email - Address',
+            'value': 'badguy@evil.com'}
+    request = client.post('/api/indicators', headers=headers, json=data)
+    response = json.loads(request.data.decode())
+    assert request.status_code == 404
+    assert response['msg'] == 'User not found by API key'
+
+    # Try an inactive API key.
+    headers = {'Authorization': 'Apikey ' + TEST_INACTIVE_APIKEY}
+    data = {'confidence': 'LOW',
+            'impact': 'LOW',
+            'status': 'New',
+            'substring': False,
+            'type': 'Email - Address',
+            'value': 'badguy@evil.com'}
+    request = client.post('/api/indicators', headers=headers, json=data)
+    response = json.loads(request.data.decode())
+    assert request.status_code == 401
+    assert response['msg'] == 'Cannot create an indicator with an inactive user'
+
+    # Try a valid API key.
+    headers = {'Authorization': 'Apikey ' + TEST_ANALYST_APIKEY}
+    data = {'confidence': 'LOW',
+            'impact': 'LOW',
+            'status': 'New',
+            'substring': False,
+            'type': 'Email - Address',
+            'value': 'badguy@evil.com'}
+    request = client.post('/api/indicators', headers=headers, json=data)
+    response = json.loads(request.data.decode())
+    assert request.status_code == 201
+    assert response['user'] == 'analyst'
+
+
 def test_create_autocreate_campaign(app, client):
     """ Ensure the auto-create campaign config actually works """