Fixes flake8 issues

harpoon/commands/robtex.py

@@ -1,13 +1,9 @@
 #! /usr/bin/env python
-import sys
-import operator
 import json
 import pytz
-from dateutil.parser import parse
 from harpoon.commands.base import Command
 from harpoon.lib.robtex import Robtex, RobtexError
 from harpoon.lib.utils import json_serial, unbracket
-from datetime import date, datetime
 
 
 class CommandRobtex(Command):
@@ -143,5 +139,3 @@ def intel(self, type, query, data):
                             )
             except RobtexError:
                 print("Robtex query failed")
-
-

harpoon/commands/subdomains.py

@@ -1,7 +1,6 @@
 #! /usr/bin/env python
 from harpoon.commands.base import Command
 from harpoon.commands.censyscmd import CommandCensys
-from censys.asm import AsmClient
 from harpoon.lib.utils import unbracket
 from passivetotal.libs.enrichment import EnrichmentRequest
 from virus_total_apis import PublicApi, PrivateApi
@@ -78,16 +77,16 @@ def prepare_data(self, subdomains=dict, data=dict, source=str):
     def intel(self, type, query, data):
         if type == "domain":
             try:
-                #subdomains = self.censys_certs(unbracket(query), self._confif_data, True)
-                #self.prepare_data(subdomains, data, "Censys")
+                # subdomains = self.censys_certs(unbracket(query), self._confif_data, True)
+                # self.prepare_data(subdomains, data, "Censys")
                 pass
 
             except CensysRateLimitExceededException:
                 print('Censys quota exceeded!')
-            #subdomains = self.pt(unbracket(query), conf, True)
-            #self.prepare_data(subdomains, data, "PassiveTotal")
-            #subdomains = self.vt(unbracket(query), conf, True)
-            #self.prepare_data(subdomains, data, "VirusTotal")
+            # subdomains = self.pt(unbracket(query), conf, True)
+            # self.prepare_data(subdomains, data, "PassiveTotal")
+            # subdomains = self.vt(unbracket(query), conf, True)
+            # self.prepare_data(subdomains, data, "VirusTotal")
 
         else:
             pass

harpoon/lib/archiveis.py

@@ -24,12 +24,15 @@ def download_cache(cache_url):
         r = requests.get(cache_url)
         data = r.text
         t1 = data.find('\n\n\n\n\n\n')
-        t2 = data.find('</div></div><!--[if !IE]><!--><div style="position:absolute;right:1028px;top:-14px;bottom:-2px">')
-        #t3 = data.find('<input style="border:1px solid black;height:20px;margin:0 0 0 0;padding:0;width:500px" type="text" name="q"')
-        #t4 = data[t3+115:].find('"')
+        t2 = data.find(
+            '</div></div><!--[if !IE]><!--><div style="position:absolute;right:1028px;' +
+            'top:-14px;bottom:-2px">')
+        # t3 = data.find('<input style="border:1px solid black;height:20px;' +
+        #  'margin:0 0 0 0;padding:0;width:500px" type="text" name="q"')
+        # t4 = data[t3+115:].find('"')
         t5 = data.find('<meta property="article:modified_time" content="')
         cached_data = data[t1+6:t2]
-        #original_url = data[t3+115:t3+115+t4]
+        # original_url = data[t3+115:t3+115+t4]
         date = parse(data[t5+48:t5+68])
         return {
             'success': True,

harpoon/lib/archiveorg.py

@@ -13,8 +13,9 @@ def snapshots(url):
         Return a list of snapshots for a given url
         """
         # FIXME: report more snapshot through the Memento API
-        r = requests.get('http://archive.org/wayback/available?url=%s' %
-                urllib.parse.quote(url))
+        r = requests.get(
+            'http://archive.org/wayback/available?url=%s' %
+            urllib.parse.quote(url))
         data = r.json()
         res = []
         if 'archived_snapshots' in data:
@@ -32,7 +33,7 @@ def download_cache(cache_url):
         Download cache from a cache url
         """
         if cache_url.startswith('https://web.archive.org/web/') or \
-            cache_url.startswith('http://web.archive.org/web/'):
+                cache_url.startswith('http://web.archive.org/web/'):
             r = requests.get(cache_url)
             data = r.text
             t1 = data.find('<!-- End Wayback Rewrite JS Include -->')

harpoon/lib/bird.py

@@ -1,5 +1,6 @@
 import tweepy
 
+
 class Bird(object):
     def __init__(self, config):
         self.consumer_key = config['consumer_key']

harpoon/lib/certspotter.py

@@ -1,6 +1,5 @@
 import requests
 from requests.auth import HTTPBasicAuth
-import json
 
 
 class CertSpotterError(Exception):
@@ -35,6 +34,10 @@ def _get(self, query, params={}):
         else:
             raise CertSpotterError("Invalid HTTP status code %i" % r.status_code)
 
-
     def search(self, domain, include_subdomains=False):
-        return self._get("v1/issuances", params = {'domain': domain, "expand": ["dns_names", "issuer", "cert"], "include_subdomains": include_subdomains})
+        return self._get(
+                "v1/issuances",
+                params={
+                    "domain": domain,
+                    "expand": ["dns_names", "issuer", "cert"],
+                    "include_subdomains": include_subdomains})

harpoon/lib/hibp.py

@@ -1,5 +1,4 @@
 import requests
-import json
 
 
 class HibpError(Exception):
@@ -26,7 +25,9 @@ def _request(self, query):
         """
         Request an url
         """
-        r = requests.get('https://haveibeenpwned.com' + query, headers= {'User-Agent': self.ua, 'hibp-api-key': self.key})
+        r = requests.get(
+            "https://haveibeenpwned.com" + query,
+            headers={'User-Agent': self.ua, 'hibp-api-key': self.key})
         if r.status_code != 200:
             if r.status_code == 404:
                 raise HibpNotFound()
@@ -52,4 +53,3 @@ def get_pastes(self, account):
 
     def check_pwd(self, pwd):
         return self._request('/api/v3/pwnedpassword/%s' % pwd).json()
-

harpoon/lib/memento.py

@@ -3,13 +3,14 @@
 from urllib.parse import urljoin, quote
 from dateutil.parser import parse
 
+
 class MementoClient(object):
     """
     Implement Memento Protocol
     """
     def __init__(self, base_url='http://archive.is/'):
         self.base_url = base_url
-        self.linkre = re.compile('^<(?P<url>[^>]+)>; rel="(?P<type>[a-z ]+)"(; datetime="(?P<date>[^"]+)"|,|; type="application/link-format"; from="(?P<from>[^"]+)"; until="(?P<until>[^"]+)")')
+        self.linkre = re.compile(r'^<(?P<url>[^>]+)>; rel="(?P<type>[a-z ]+)"(; datetime="(?P<date>[^"]+)"|,|; type="application/link-format"; from="(?P<from>[^"]+)"; until="(?P<until>[^"]+)")')  # noqa: E501
 
     def _parselinks(self, data):
         """
@@ -56,7 +57,7 @@ def snapshots(self, url):
                 if d['type'] in ['memento', 'first last memento']:
                     snapshots.append({
                         'url': original,
-                        'date':d['date'],
+                        'date': d['date'],
                         'archive': d['url']
                     })
             return snapshots

harpoon/lib/pgp.py

@@ -14,7 +14,7 @@ def search(search):
             return []
         res = []
         soup = BeautifulSoup(r.text, 'lxml')
-        emailsearch = re.compile("([\w\(\) ]+) <([\w@\.]+)>")
+        emailsearch = re.compile(r"([\w\(\) ]+) <([\w@\.]+)>")
         for pre in soup.find_all('pre')[1:]:
             key = {}
             t = str(pre)

harpoon/lib/pulsedive.py

@@ -6,6 +6,7 @@ def __init__(self, message):
         Exception.__init__(self, message)
         self.message = message
 
+
 class PulseDiveNotFound(PulseDiveError):
     pass
 
@@ -37,7 +38,7 @@ def indicators_by_value(self, val, historical=False):
         information is retrieved using the indicator value. Querying
         links and properties by indicator value won't work.
         """
-        params  = {'indicator': val}
+        params = {'indicator': val}
         if historical:
             params['historical'] = "1"
         return self._query("info.php", params)
@@ -53,6 +54,3 @@ def threat(self, threat):
 
     def threat_indicators(self, tid):
         return self._query("info.php", {'tid': tid, 'get': 'links'})
-
-
-

harpoon/lib/urlhaus.py

@@ -1,7 +1,6 @@
 #! /usr/bin/env python3
 
 import re
-
 import requests
 
 
@@ -57,7 +56,7 @@ def download_sample(self, url, query):
             if r.ok:
                 try:
                     return r.content
-                except:
+                except:  # noqa: E722
                     raise UrlHausError()
             else:
                 return None
@@ -85,9 +84,9 @@ def get_payload(self, query):
         """
         url = "{}payload/".format(self.base_url)
         # check if the hash is md5 or sha256
-        if re.match("[A-Fa-f0-9]{64}", query.strip()):
+        if re.match(r"[A-Fa-f0-9]{64}", query.strip()):
             return self.launch_post_query(url, "sha256_hash", query.strip())
-        elif re.match("[a-fA-F\d]{32}", query.strip()):
+        elif re.match(r"[a-fA-F\d]{32}", query.strip()):
             return self.launch_post_query(url, "md5_hash", query.strip())
         else:
             raise UrlHausError()

harpoon/lib/xforce.py

@@ -1,6 +1,5 @@
 import requests
 import base64
-import json
 
 
 class XforceExchangeFailed(Exception):
@@ -99,7 +98,3 @@ def whois(self, domain):
         Returns a JSON object containing information about the given host address.
         """
         return self._request('/whois/' + domain)
-
-
-
-

tests/command/test_circl.py

@@ -1,5 +1,3 @@
-import pytest
-
 from harpoon.commands.circl import CommandCircl
 from ..utils import launch_plugin
 

tests/command/test_crtsh.py

@@ -1,5 +1,3 @@
-import pytest
-
 from harpoon.commands.crtsh import CommandCertsh
 from ..utils import launch_plugin
 

tests/command/test_dnsdb.py

@@ -1,5 +1,3 @@
-import pytest
-
 from harpoon.commands.dnsdbcmd import DnsDbTotal
 from ..utils import launch_plugin
 

tests/command/test_hashlookup.py

@@ -1,5 +1,3 @@
-import pytest
-
 from harpoon.commands.hashlookup import CommandHashLookup
 from ..utils import launch_plugin
 

tests/command/test_hybrid.py

@@ -1,5 +1,3 @@
-import pytest
-
 from harpoon.commands.hybrid import CommandHybridAnalysis
 from ..utils import launch_plugin
 

tests/command/test_otx.py

@@ -1,5 +1,3 @@
-import pytest
-
 from harpoon.commands.otx import CommandOtx
 from ..utils import launch_plugin
 

tests/command/test_quad9.py

@@ -1,5 +1,3 @@
-import pytest
-
 from harpoon.commands.quad9 import CommandQuad9
 from ..utils import launch_plugin
 

tests/command/test_safebrowsing.py

@@ -1,5 +1,3 @@
-import pytest
-
 from harpoon.commands.safebrowsing import CommandSafeBrowsing
 from ..utils import launch_plugin
 

tests/test_utils.py

@@ -0,0 +1,20 @@
+from harpoon.lib.utils import unbracket, bracket, same_url, typeguess
+
+
+class TestUtils:
+    def test_bracket(self):
+        assert unbracket("domain[.]com") == "domain.com"
+        assert bracket("domain.com") == "domain[.]com"
+
+    def test_same_url(self):
+        assert same_url("http://example.org", "https://example.org")
+        assert same_url("http://example.org", "http://www.example.org")
+        assert same_url("http://exemple.org", "http://example.org") is False
+
+    def test_typeguess(self):
+        assert typeguess("44c13fc77ee90ef4040a0c99a9be999e") == "md5"
+        assert typeguess("d82b0fffdda6d7120dd8c14da32208278e2a287f") == "sha1"
+        assert typeguess("fef75dbbf6297c151a7112cb5f98884e4928716f0725826c42086e6c21a9894d") == "sha256"
+        assert typeguess("10.4.4.4") == "IPv4"
+        assert typeguess("fce8::1") == "IPv6"
+        assert typeguess("domain.com") == "domain"