feature/normalize fqdns (csirtgadgets#148)

* Add test-cases for normalizing urls/fqdns

* Add matching for dots at end of fqdn

And create `normalize_indicator` func

* Implement `normalize_indicator` func for urls/fqdns

* Fix imports

* Init github action for commit tests

* Relocate .github dir to root

* Add tests on PRs

* Don't lowercase `reference` field

Author: mdavis332

.github/workflows/push_test.yml

@@ -0,0 +1,27 @@
+name: Test latest commit
+
+on: [push, pull_request]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Set up Python 3.6
+        uses: actions/setup-python@v1
+        with:
+          python-version: 3.6
+      
+      - uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+      
+      - name: Install dependencies
+        run: |
+          sudo apt-get update
+          
+          python -m pip install --upgrade pip
+          pip install -r dev_requirements.txt
+          pip install stix
+          python setup.py test
+          python setup.py sdist bdist bdist_wheel

csirtg_indicator/indicator.py

@@ -6,7 +6,8 @@
 from base64 import b64encode
 from .exceptions import InvalidIndicator
 from . import VERSION
-from .utils import parse_timestamp, resolve_itype, is_subdomain, ipv4_normalize
+from .utils import resolve_itype, is_subdomain, ipv4_normalize, normalize_indicator
+from .utils.ztime import parse_timestamp
 import pytricia
 import codecs
 from datetime import datetime
@@ -33,7 +34,7 @@ def __init__(self, indicator=None, version=PROTOCOL_VERSION, **kwargs):
         self.version = version
         if 'lowercase' in kwargs:
             self._lowercase = kwargs.get('lowercase')
-            # indicate lowercase arg was explicitly passed by user rather than just a  default value
+            # indicate lowercase arg was explicitly passed by user rather than just a default value
             self._lowercase_explicit = True
         else:
             # set lowercase to True by default, but ensure we can later determine it was not user specified
@@ -55,10 +56,10 @@ def __init__(self, indicator=None, version=PROTOCOL_VERSION, **kwargs):
                 continue
 
             if isinstance(kwargs[k], basestring):
-                # always stripe whitespace
+                # always strip whitespace
                 kwargs[k] = kwargs[k].strip()
 
-                if self._lowercase is True:
+                if self._lowercase is True and k != 'reference': # don't lower reference which may be a url
                     kwargs[k] = kwargs[k].lower()
                 if k in ['tags', 'peers']:
                     kwargs[k] = kwargs[k].split(',')
@@ -115,20 +116,16 @@ def indicator(self, i):
         self.itype = resolve_itype(i.lower())
         self._indicator = i
 
-        if self.itype == 'url':
-            u = urlparse(self._indicator)
-            if self._lowercase is True and self._lowercase_explicit is True:
-                self._indicator = u.geturl().rstrip('/').lower()
-            else:
-                self._indicator = u.geturl().rstrip('/')
-        else:
-            if self._lowercase is True:
-                self._indicator = self._indicator.lower()
-
+        if self.itype in ['url', 'fqdn']:
+            self._indicator = normalize_indicator(self._indicator, itype=self.itype, 
+                lowercase=self._lowercase, lowercase_explicit=self._lowercase_explicit)
 
-        if self.itype == 'ipv4':
+        elif self.itype == 'ipv4':
             self._indicator = ipv4_normalize(self._indicator)
 
+        else:
+            self._indicator = self._indicator.lower()
+
         if self.mask and (self.itype in ['ipv4', 'ipv6']):
             self._indicator = '{}/{}'.format(self._indicator, int(self.mask))
             self.mask = None
@@ -297,7 +294,7 @@ def __repr__(self):
                 v = v.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
 
             if isinstance(v, basestring):
-                if k not in ['indicator', 'message'] and not k.endswith('time') and self._lowercase is True:
+                if k not in ['indicator', 'message', 'reference'] and not k.endswith('time') and self._lowercase is True:
                     v = v.lower()
 
             if k == 'confidence':

csirtg_indicator/utils/__init__.py

@@ -3,41 +3,41 @@
 import ipaddress
 from ..exceptions import InvalidIndicator
 from ..constants import PYVERSION
-from .ztime import parse_timestamp
-import sys
+
 
 if PYVERSION == 3:
     from urllib.parse import urlparse
+    from urllib.parse import urlsplit
+    from urllib.parse import urlunsplit
 else:
     from urlparse import urlparse
 
-from pprint import pprint
 
-RE_IPV4 = re.compile('^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(\d{1,3})$')
-RE_IPV4_CIDR = re.compile('^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\/\d{1,2})$')
+RE_IPV4 = re.compile(r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(\d{1,3})$')
+RE_IPV4_CIDR = re.compile(r'^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\/\d{1,2})$')
 
 # http://stackoverflow.com/a/17871737
-RE_IPV6 = re.compile('(([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(:[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(ffff(:0{1,4}){0,1}:){0,1}((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])|([0-9a-fA-F]{1,4}:){1,4}:((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))')
+RE_IPV6 = re.compile(r'(([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(:[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(ffff(:0{1,4}){0,1}:){0,1}((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])|([0-9a-fA-F]{1,4}:){1,4}:((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))')
 
 # http://goo.gl/Cztyn2 -- probably needs more work
 # http://stackoverflow.com/a/26987741/7205341
 # ^((xn--)?(--)?[a-zA-Z0-9-_@]+(-[a-zA-Z0-9]+)*\.)+[a-zA-Z]{2,}(--p1ai)?$
 #RE_FQDN = re.compile('^((?!-))(xn--)?[a-z0-9][a-z0-9-_\.]{0,61}[a-z0-9]{0,1}\.(xn--)?([a-z0-9\-]{1,61}|[a-z0-9-]{1,30}\.[a-z]{2,})$')
 # http://stackoverflow.com/questions/14402407/maximum-length-of-a-domain-name-without-the-http-www-com-parts
-RE_FQDN = re.compile('^((?!-))(xn--)?[a-z0-9][a-z0-9-_\.]{0,245}[a-z0-9]{0,1}\.(xn--)?([a-z0-9\-]{1,61}|[a-z0-9-]{1,30}\.[a-z]{2,})$')
-RE_URI_SCHEMES = re.compile('^(https?|ftp)$')
-RE_EMAIL = re.compile("^[-\w+.!#$%&'*\/=?^_`{|}~;]+@[-.0-9a-zA-Z][-.0-9a-zA-Z]*[a-zA-Z]{2,}$")
-RE_ASN = re.compile('^(AS|as)[0-9]{1,6}$')
+RE_FQDN = re.compile(r'^((?!-))(xn--)?[a-z0-9][a-z0-9-_\.]{0,245}[a-z0-9]{0,1}\.(xn--)?([a-z0-9\-]{1,61}|[a-z0-9-]{1,30}\.[a-z]{2,})\.?$')
+RE_URI_SCHEMES = re.compile(r'^(https?|ftp)$')
+RE_EMAIL = re.compile(r"^[-\w+.!#$%&'*\/=?^_`{|}~;]+@[-.0-9a-zA-Z][-.0-9a-zA-Z]*[a-zA-Z]{2,}$")
+RE_ASN = re.compile(r'^(AS|as)[0-9]{1,6}$')
 
 RE_HASH = {
-    'uuid': re.compile('^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$'),
-    'md5': re.compile('^[a-fA-F0-9]{32}$'),
-    'sha1': re.compile('^[a-fA-F0-9]{40}$'),
-    'sha256': re.compile('^[a-fA-F0-9]{64}$'),
-    'sha512': re.compile('^[a-fA-F0-9]{128}$'),
+    'uuid': re.compile(r'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$'),
+    'md5': re.compile(r'^[a-fA-F0-9]{32}$'),
+    'sha1': re.compile(r'^[a-fA-F0-9]{40}$'),
+    'sha256': re.compile(r'^[a-fA-F0-9]{64}$'),
+    'sha512': re.compile(r'^[a-fA-F0-9]{128}$'),
 }
 
-RE_IPV4_PADDING = re.compile(r"(^|\.)0+([^/.])")
+RE_IPV4_PADDING = re.compile(r'(^|\.)0+([^/.])')
 
 
 def ipv4_normalize(i):
@@ -205,6 +205,25 @@ def normalize_itype(i, itype=None):
     return i
 
 
+def normalize_indicator(i, itype=None, lowercase=False, lowercase_explicit=False):
+    if itype == 'fqdn':
+        i = i.rstrip('.')
+        # only don't lowercase if lowercase=False and lowercase_explicit=True (set by user)
+        if lowercase or not lowercase_explicit:
+            i = i.lower()
+    elif itype == 'url':
+        u = urlparse(i)
+        i = u.geturl().rstrip('/')
+        if lowercase and lowercase_explicit:
+            i = i.lower()
+        elif lowercase or not lowercase_explicit:
+            scheme, netloc, path, qs, anchor = urlsplit(i)
+            netloc = netloc.rstrip('.').lower()
+            i = urlunsplit((scheme, netloc, path, qs, anchor))
+
+    return i
+
+
 def is_subdomain(i):
     itype = resolve_itype(i)
     if itype is not 'fqdn':

test/test_fqdn.py

@@ -1,5 +1,4 @@
 from csirtg_indicator import Indicator
-from csirtg_indicator.utils import is_subdomain
 from faker import Faker
 fake = Faker()
 
@@ -12,6 +11,8 @@
     'xn----jtbbmekqknepg3a.xn--p1ai',
     'dualstack.cddf-prod-frontend-1ho73vqwbi0tw-1326553765.us-east-1.elb.amazonaws.com',
     'laser-retargeting-server-production.us-east-1-prod-core-edge-public.spongecell.net',
+    'example.org.',
+    'an0ther.exAmple.orG.',
 ]
 
 
@@ -39,14 +40,17 @@ def test_fqdn_urls():
 def test_fqdn_ok():
 
     for d in GOOD:
-        d = Indicator(d)
-        assert d.itype is 'fqdn'
+        e = Indicator(d)
+        assert e.itype is 'fqdn'
+        d = d.rstrip('.')
+        assert e.indicator == d.lower()
 
 
 def test_fqdn_subdomain():
     data = [
         'www.yahoo.com',
         'www.ww2.yahoo.com',
+        'this.is.aNother.sub.domain.tld.',
     ]
 
     for d in data:
@@ -57,6 +61,7 @@ def test_fqdn_subdomain():
     data = [
         'yahoo.com',
         'google.com',
+        'notasubdomain.tLd.',
         'http://google.com',
         'https://www.google.com',
         'http://www2.www.google.com',

test/test_indicator.py

@@ -1,3 +1,4 @@
+from weakref import ref
 from csirtg_indicator import Indicator
 import json
 from csirtg_indicator.exceptions import InvalidIndicator
@@ -30,9 +31,10 @@ def test_indicator_url():
     assert 'malware' in i.tags
 
 
-def test_indicator_mixedcase_lower_false():
-    i = Indicator('http://example.org/MiXeDCaSe',
-                  tags='botnet,malware', lowercase=False)
+# by default, if it's a url, lower the hostname but not the path
+def test_indicator_mixedcase_lower_default():
+    i = Indicator('http://exAmple.Org./MiXeDCaSe',
+                  tags='botnet,malware')
 
     assert i.is_private() is False
     assert i.indicator == 'http://example.org/MiXeDCaSe'
@@ -41,7 +43,19 @@ def test_indicator_mixedcase_lower_false():
     assert 'botnet' in i.tags
     assert 'malware' in i.tags
 
+# if it's a url and user explicitly says don't lowercase, leave entire indicator as-is
+def test_indicator_mixedcase_lower_false():
+    i = Indicator('http://examPle.org/MiXeDCaSe',
+                  tags='botnet,malware', lowercase=False)
+
+    assert i.is_private() is False
+    assert i.indicator == 'http://examPle.org/MiXeDCaSe'
+    assert i.itype is not 'fqdn'
+    assert i.itype is 'url'
+    assert 'botnet' in i.tags
+    assert 'malware' in i.tags
 
+# if it's a url and user explicitly says to lowercase, force all parts to lowercase
 def test_indicator_mixedcase_lower_true():
     i = Indicator('http://example.org/MiXeDCaSe',
                   tags='botnet,malware', lowercase=True)
@@ -136,3 +150,12 @@ def test_eq():
 
     u2.uuid = u1.uuid
     assert u1 == u2
+
+def test_reference_field_case():
+    ref_url = 'http://good.intel.tld/API/Generator'
+    i = Indicator('http://example.org', tags='botnet,malware', reference=ref_url)
+
+    assert i.reference == ref_url
+
+    x = json.loads(i.__repr__())
+    assert x['reference'] == ref_url

test/test_urls.py

@@ -15,6 +15,7 @@
     'http://[email protected]/aol5/a000l.html',
     'https://example.com:443/1.html',
     'http://test1.test2.example.com',
+    'http://tEst1.test3.exAmple.com.',
     'http://xz.job391.com/down/����࿪��@89_1_60',
     'http://refreshdharan.com/bg/excel2/index.php?userid={[email protected]}',
     'http://https.www.paypal.blahblahblahblah.web.cgi.bin.blahblah.blahblahblahblah.blahblahblah-blah-blah-blah.com/signin/',