Merge pull request #50 from automationator/master

Couple updates

docker-compose-DEV.yml

@@ -30,7 +30,6 @@ services:
             - mysql-dev:/var/lib/mysql
         env_file:
             - ./services/db/docker-DEV.env
-        #command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_cs --explicit_defaults_for_timestamp
         command: --explicit_defaults_for_timestamp
 
     nginx-dev:

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

@@ -4,7 +4,7 @@
 
 from dateutil.parser import parse
 from flask import current_app, jsonify, request, Response, url_for
-from sqlalchemy import and_, exc
+from sqlalchemy import and_, exc, func
 
 from project import db
 from project.api import bp
@@ -157,17 +157,22 @@ def create_indicator():
         else:
             return error_response(404, 'Indicator type not found: {}'.format(data['type']))
 
-    # Verify this type+value does not already exist.
-    existing = Indicator.query.filter_by(type=indicator_type, value=data['value']).first()
-    if existing:
-        return error_response(409, 'Indicator already exists')
-
     # Verify the case-sensitive value (defaults to False).
     if 'case_sensitive' in data:
         case_sensitive = data['case_sensitive']
     else:
         case_sensitive = False
 
+    # Verify this type+value does not already exist based off of case_sensitive.
+    if case_sensitive:
+        existing = Indicator.query.filter(Indicator.type == indicator_type, func.binary(Indicator.value) == func.binary(data['value'])).first()
+        if existing:
+            return error_response(409, 'Case-sensitive indicator already exists')
+    else:
+        existing = Indicator.query.filter(Indicator.type == indicator_type, func.lower(Indicator.value) == func.lower(data['value'])).first()
+        if existing:
+            return error_response(409, 'Case-insensitive indicator already exists')
+
     # Verify the confidence (has default).
     if 'confidence' not in data:
         confidence = IndicatorConfidence.query.order_by(IndicatorConfidence.id).limit(1).first()

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

@@ -317,7 +317,11 @@ def test_create_duplicate(client):
 
     request, response = create_indicator(client, 'asdf', 'asdf', 'analyst')
     assert request.status_code == 409
-    assert response['msg'] == 'Indicator already exists'
+    assert response['msg'] == 'Case-insensitive indicator already exists'
+
+    request, response = create_indicator(client, 'asdf', 'ASDF', 'analyst')
+    assert request.status_code == 409
+    assert response['msg'] == 'Case-insensitive indicator already exists'
 
 
 def test_create_nonexistent_username(client):
@@ -635,6 +639,51 @@ def test_create_autocreate_tag(app, client):
     assert response['tags'] == ['from_address', 'phish']
 
 
+def test_create_case_sensitivity(client):
+    """ Ensure the database case-sensitivity is working """
+
+    request, response = create_indicator(client, 'asdf', 'asdf', 'analyst',
+                                         campaigns=['LOLcats', 'Derpsters'],
+                                         case_sensitive=False,
+                                         confidence='HIGH',
+                                         impact='HIGH',
+                                         intel_reference='http://blahblah.com',
+                                         intel_source='OSINT',
+                                         status='Analyzed',
+                                         substring=False,
+                                         tags=['phish', 'nanocore'])
+    assert request.status_code == 201
+    assert response['type'] == 'asdf'
+    assert response['value'] == 'asdf'
+
+    request, response = create_indicator(client, 'asdf', 'ASDF', 'analyst',
+                                         campaigns=['LOLcats', 'Derpsters'],
+                                         case_sensitive=False,
+                                         confidence='HIGH',
+                                         impact='HIGH',
+                                         intel_reference='http://blahblah.com',
+                                         intel_source='OSINT',
+                                         status='Analyzed',
+                                         substring=False,
+                                         tags=['phish', 'nanocore'])
+    assert request.status_code == 409
+    assert response['msg'] == 'Case-insensitive indicator already exists'
+
+    request, response = create_indicator(client, 'asdf', 'ASDF', 'analyst',
+                                         campaigns=['LOLcats', 'Derpsters'],
+                                         case_sensitive=True,
+                                         confidence='HIGH',
+                                         impact='HIGH',
+                                         intel_reference='http://blahblah.com',
+                                         intel_source='OSINT',
+                                         status='Analyzed',
+                                         substring=False,
+                                         tags=['phish', 'nanocore'])
+    assert request.status_code == 201
+    assert response['type'] == 'asdf'
+    assert response['value'] == 'ASDF'
+
+
 def test_create(client):
     """ Ensure a proper request actually works """