Merge pull request #1483 from p-l-/black-is-the-new-black

Linting updates (Black & Pylint)

.github/workflows/linting.yml

@@ -1,5 +1,5 @@
 # This file is part of IVRE.
-# Copyright 2011 - 2022 Pierre LALET <pierre@droids-corp.org>
+# Copyright 2011 - 2023 Pierre LALET <pierre@droids-corp.org>
 #
 # IVRE is free software: you can redistribute it and/or modify it
 # under the terms of the GNU General Public License as published by
@@ -50,7 +50,7 @@ jobs:
 
     - run: git ls-files | grep -vE '^web/static/(doc|an|bs|d3|jq|lk)/|^data/|\.(png|gif|svg)$' | xargs -r codespell --ignore-words=pkg/codespell_ignore
 
-    - run: pylint -e all -d abstract-method,arguments-differ,attribute-defined-outside-init,broad-except,duplicate-code,fixme,function-redefined,global-statement,import-error,invalid-name,locally-disabled,missing-docstring,no-member,protected-access,super-init-not-called,suppressed-message,too-few-public-methods,too-many-ancestors,too-many-arguments,too-many-boolean-expressions,too-many-branches,too-many-instance-attributes,too-many-lines,too-many-locals,too-many-nested-blocks,too-many-public-methods,too-many-return-statements,too-many-statements,unsubscriptable-object,unused-argument,line-too-long,consider-using-f-string ivre ./doc/conf.py ./setup.py ./bin/ivre
+    - run: pylint -e all -d abstract-method,arguments-differ,attribute-defined-outside-init,broad-except,broad-exception-raised,duplicate-code,fixme,function-redefined,global-statement,import-error,invalid-name,locally-disabled,missing-docstring,no-member,protected-access,super-init-not-called,suppressed-message,too-few-public-methods,too-many-ancestors,too-many-arguments,too-many-boolean-expressions,too-many-branches,too-many-instance-attributes,too-many-lines,too-many-locals,too-many-nested-blocks,too-many-public-methods,too-many-return-statements,too-many-statements,unsubscriptable-object,unused-argument,use-dict-literal,line-too-long,consider-using-f-string ivre ./doc/conf.py ./setup.py ./bin/ivre
     - run: pylint -e all -d unused-argument,too-many-arguments,missing-function-docstring,missing-class-docstring,missing-module-docstring,multiple-statements,invalid-name,too-few-public-methods ./pkg/stubs/*.pyi
 
     - run: isort --profile black --check-only ivre ./doc/conf.py ./setup.py ./bin/ivre ./tests/tests.py ./pkg/stubs/*.pyi

ivre/analyzer/ntlm.py

@@ -2,7 +2,7 @@
 # -*- coding: utf-8 -*-
 
 # This file is part of IVRE.
-# Copyright 2011 - 2021 Pierre LALET <pierre@droids-corp.org>
+# Copyright 2011 - 2023 Pierre LALET <pierre@droids-corp.org>
 #
 # IVRE is free software: you can redistribute it and/or modify it
 # under the terms of the GNU General Public License as published by
@@ -329,14 +329,13 @@ def _is_ntlm_message(message: str) -> bool:
         val1, val2 = message.split(None, 1)
     except ValueError:
         return False
-    else:
-        if not val2:
-            return False
-        if val1.lower() == "ntlm":
-            return True
-        if val1.lower() == "negotiate":
-            try:
-                return utils.decode_b64(val2.encode())[:7] == b"NTLMSSP"
-            except (UnicodeDecodeError, TypeError, ValueError, binascii.Error):
-                pass
+    if not val2:
+        return False
+    if val1.lower() == "ntlm":
+        return True
+    if val1.lower() == "negotiate":
+        try:
+            return utils.decode_b64(val2.encode())[:7] == b"NTLMSSP"
+        except (UnicodeDecodeError, TypeError, ValueError, binascii.Error):
+            pass
     return False

ivre/db/__init__.py

@@ -848,7 +848,6 @@ def display_top(self, arg, flt, lmt):
 
 
 class DBActive(DB):
-
     ipaddr_fields = ["addr", "traces.hops.ipaddr", "ports.state_reason_ip"]
     datetime_fields = [
         "starttime",
@@ -2112,7 +2111,6 @@ def cmp_schema_version_scan(_):
 
 
 class DBNmap(DBActive):
-
     backends = {
         "http": ("http", "HttpDBNmap"),
         "mongodb": ("mongo", "MongoDBNmap"),
@@ -3790,7 +3788,6 @@ def store_scan_json_dismap(
 
 
 class DBView(DBActive):
-
     backends = {
         "elastic": ("elastic", "ElasticDBView"),
         "http": ("http", "HttpDBView"),
@@ -3956,7 +3953,6 @@ def update(self, timestamp):
 
 
 class DBPassive(DB):
-
     argparser_insert = ArgumentParser(add_help=False)
     ipaddr_fields = ["addr"]
     datetime_fields = ["firstseen", "lastseen", "infos.not_after", "infos.not_before"]
@@ -4403,7 +4399,6 @@ def searchhassh(cls, value_or_hash=None, server=None):
 
 
 class DBData(DB):
-
     country_codes = None
     backends = {
         "http": ("http", "HttpDBData"),
@@ -4857,10 +4852,10 @@ class DBFlowMeta(type):
     instances of MongoDBFlow and TinyDBFlow.
     """
 
-    def __new__(cls, name, bases, attrs):
+    def __new__(mcs, name, bases, attrs):
         attrs["meta_desc"] = DBFlowMeta.compute_meta_desc()
         attrs["list_fields"] = DBFlowMeta.compute_list_fields(attrs["meta_desc"])
-        return type.__new__(cls, name, bases, attrs)
+        return type.__new__(mcs, name, bases, attrs)
 
     @staticmethod
     def compute_meta_desc():
@@ -5314,7 +5309,6 @@ def topvalues(
 
 
 class MetaDB:
-
     db_types = {
         "nmap": DBNmap,
         "view": DBView,

ivre/db/elastic.py

@@ -38,7 +38,6 @@
 
 
 class ElasticDB(DB):
-
     nested_fields = []
 
     # filters
@@ -231,7 +230,6 @@ def _create_mappings(nested, all_mappings):
 
 
 class ElasticDBActive(ElasticDB, DBActive):
-
     nested_fields = [
         "ports",
         "ports.scripts",

ivre/db/http.py

@@ -193,7 +193,6 @@ def pincode(self):
 
 
 class HttpDB(DB):
-
     flt_empty = {}
     no_limit = None
 

ivre/db/maxmind.py

@@ -315,7 +315,6 @@ def get_ranges(self, fields, cond=None):
 
 
 class MaxMindDBData(DBData):
-
     LANG = "en"
 
     AS_KEYS = {

ivre/db/mongo.py

@@ -140,7 +140,6 @@ def db(self):
 
 
 class MongoDB(DB):
-
     is_documentdb = False  # set to True for AWS DocumentDB sub-classes
     indexes: List[List[Tuple[List[IndexKey], Dict[str, Any]]]] = []
     schema_migrations_indexes: List[
@@ -809,7 +808,6 @@ def searchtext(text):
 
 
 class MongoDBActive(MongoDB, DBActive):
-
     column_hosts = 0
     _features_column = 0
     indexes: List[List[Tuple[List[IndexKey], Dict[str, Any]]]] = [
@@ -4379,7 +4377,6 @@ def categories_to_val(categories):
 
 
 class MongoDBNmap(MongoDBActive, DBNmap):
-
     column_scans = 1
     content_handler = Nmap2Mongo
 
@@ -4499,7 +4496,6 @@ def store_or_merge_host(self, host):
 
 
 class MongoDBPassive(MongoDB, DBPassive):
-
     column_passive = 0
     _features_column = 0
     indexes = [

ivre/db/sql/__init__.py

@@ -278,7 +278,6 @@ def fixline(self, line):
 
 
 class SQLDB(DB):
-
     table_layout = namedtuple("empty_layout", [])
     tables = table_layout()
     fields = {}
@@ -2605,8 +2604,8 @@ def is_scan_present(self, scanid):
     def searchsource(cls, src, neg=False):
         if isinstance(src, list):
             if neg:
-                return cls.base_filter(main=(cls.tables.scan.source.notin_(src)))
-            return cls.base_filter(main=(cls.tables.scan.source.in_(src)))
+                return cls.base_filter(main=cls.tables.scan.source.notin_(src))
+            return cls.base_filter(main=cls.tables.scan.source.in_(src))
         return cls.base_filter(
             main=cls._searchstring_re(cls.tables.scan.source, src, neg=neg)
         )
@@ -3214,7 +3213,7 @@ def searchrecontype(cls, rectype, neg=False):
                 return PassiveFilter(
                     main=(cls.tables.passive.recontype.notin_(rectype))
                 )
-            return PassiveFilter(main=(cls.tables.passive.recontype.in_(rectype)))
+            return PassiveFilter(main=cls.tables.passive.recontype.in_(rectype))
         return PassiveFilter(
             main=cls._searchstring_re(cls.tables.passive.recontype, rectype, neg=neg)
         )

ivre/db/sql/postgres.py

@@ -747,7 +747,9 @@ def topvalues(
             field = self._topstructure(
                 self.tables.script,
                 [
-                    func.jsonb_array_elements(self.tables.script.data["ssl-cert"],).op(
+                    func.jsonb_array_elements(
+                        self.tables.script.data["ssl-cert"],
+                    ).op(
                         "->" if subfield in ["subject", "issuer", "pubkey"] else "->>"
                     )(subfield)
                 ],
@@ -1993,6 +1995,7 @@ def insert_or_update_bulk(
             )
             self.db.execute(delete(tmp))
             if config.DEBUG_DB:
+                # pylint: disable=used-before-assignment,undefined-variable
                 stop_time = time.time()
                 time_spent = stop_time - start_time
                 total_upserted += count_upserted

ivre/db/sql/tables.py

@@ -122,7 +122,6 @@ def extend_binary_expression(element, compiler, **kwargs):
 
 
 class DefaultJSONB(UserDefinedType):
-
     python_type = dict
     cache_ok = False
 
@@ -155,7 +154,6 @@ def process(value):
 
 
 class DefaultARRAY(TypeDecorator):
-
     impl = Text
     cache_ok = False
 
@@ -181,7 +179,6 @@ def SQLARRAY(item_type):
 
 
 class DefaultINET(UserDefinedType):
-
     python_type = bytes
     cache_ok = True
 
@@ -209,7 +206,6 @@ def process(value):
 
 
 class Point(UserDefinedType):
-
     cache_ok = True
 
     def get_col_spec(self):

ivre/db/tiny.py

@@ -110,7 +110,6 @@ def count(self, flt):
     def _db_get(self, flt, fields=None, sort=None, limit=None, skip=None):
         result = self.db.search(flt)
         if fields is not None:
-
             _fields = {}
             for fld in fields:
                 try:
@@ -158,7 +157,7 @@ def _extractor(rec, wanted_fields, base=""):
             return result
 
         def _cmp(v1, v2):
-            for (k, o) in sort:
+            for k, o in sort:
                 f1 = v1
                 f2 = v2
                 for sk in k.split("."):

ivre/tools/localscan.py

@@ -37,7 +37,6 @@
 
 
 class LocalPorts:
-
     cmd_addrs = ["ip", "addr", "show"]
     cmd_openports = ["ss", "-ltunp"]
     match_addrs = re.compile("^ +inet (?P<addr>[0-9\\.]+)/")

ivre/tools/scancli.py

@@ -218,7 +218,6 @@ def displayfunction(cur: DBCursor) -> None:
             )
 
     elif args.to_db is not None:
-
         outdb = DBNmap.from_url(args.to_db)
 
         def displayfunction(cur: DBCursor) -> None:

ivre/tools/view.py

@@ -223,7 +223,6 @@ def displayfunction(cur: DBCursor) -> None:
             )
 
     elif args.to_db is not None:
-
         outdb = DBView.from_url(args.to_db)
 
         def displayfunction(cur: DBCursor) -> None:

ivre/utils.py

@@ -1582,9 +1582,9 @@ def parse_line(line: str) -> None:
     _WIRESHARK_MANUF_DB_POPULATED = True
 
 
-def get_wireshark_manuf_db() -> Tuple[
-    List[int], List[Optional[Tuple[str, Optional[str]]]]
-]:
+def get_wireshark_manuf_db() -> (
+    Tuple[List[int], List[Optional[Tuple[str, Optional[str]]]]]
+):
     if not _WIRESHARK_MANUF_DB_POPULATED:
         _read_wireshark_manuf_db()
     return _WIRESHARK_MANUF_DB_LAST_ADDR, _WIRESHARK_MANUF_DB_VALUES

ivre/web/app.py

@@ -69,7 +69,6 @@ def _die(referer):
 
     @wraps(func)
     def _newfunc(*args, **kargs):
-
         # Header with an existing X-API-Key header or an
         # Authorization: Bearer XXX are OK. Note: IVRE does not check
         # those values, they only serve as anti-CSRF protections.

ivre/web/utils.py

@@ -261,7 +261,7 @@ def add_unused(neg, param, value):
             )
         )
 
-    for (neg, param, value) in query:
+    for neg, param, value in query:
         if not neg and param == "skip":
             skip = int(value)
         elif not neg and param == "limit":

pkg/runchecks

@@ -1,7 +1,7 @@
 #! /bin/bash
 
 # This file is part of IVRE.
-# Copyright 2011 - 2022 Pierre LALET <pierre@droids-corp.org>
+# Copyright 2011 - 2023 Pierre LALET <pierre@droids-corp.org>
 #
 # IVRE is free software: you can redistribute it and/or modify it
 # under the terms of the GNU General Public License as published by
@@ -31,7 +31,7 @@ MYPYPATH=./pkg/stubs/ mypy --follow-imports=skip --disallow-untyped-calls --disa
 
 git ls-files | grep -vE '^web/static/(doc|an|bs|d3|jq|lk)/|^data/|\.(png|gif|svg)$' | xargs -r codespell --ignore-words=pkg/codespell_ignore && echo "codespell OK"
 
-pylint -e all -d abstract-method,arguments-differ,attribute-defined-outside-init,broad-except,duplicate-code,fixme,function-redefined,global-statement,import-error,invalid-name,locally-disabled,missing-docstring,no-member,protected-access,super-init-not-called,suppressed-message,too-few-public-methods,too-many-ancestors,too-many-arguments,too-many-boolean-expressions,too-many-branches,too-many-instance-attributes,too-many-lines,too-many-locals,too-many-nested-blocks,too-many-public-methods,too-many-return-statements,too-many-statements,unsubscriptable-object,unused-argument,line-too-long,consider-using-f-string ivre ./doc/conf.py ./setup.py ./bin/ivre && echo "pylint OK"
+pylint -e all -d abstract-method,arguments-differ,attribute-defined-outside-init,broad-except,broad-exception-raised,duplicate-code,fixme,function-redefined,global-statement,import-error,invalid-name,locally-disabled,missing-docstring,no-member,protected-access,super-init-not-called,suppressed-message,too-few-public-methods,too-many-ancestors,too-many-arguments,too-many-boolean-expressions,too-many-branches,too-many-instance-attributes,too-many-lines,too-many-locals,too-many-nested-blocks,too-many-public-methods,too-many-return-statements,too-many-statements,unsubscriptable-object,unused-argument,use-dict-literal,line-too-long,consider-using-f-string ivre ./doc/conf.py ./setup.py ./bin/ivre && echo "pylint OK"
 
 pylint -e all -d unused-argument,too-many-arguments,missing-function-docstring,missing-class-docstring,missing-module-docstring,multiple-statements,invalid-name,too-few-public-methods ./pkg/stubs/*.pyi && echo "pylint stubs OK"
 

tests/tests.py

@@ -302,7 +302,6 @@ def scan(self, target_options):
 
 
 class IvreTests(unittest.TestCase):
-
     maxDiff = None
 
     def setUp(self):
@@ -848,7 +847,6 @@ def test_20_fake_nmap_passive(self):
         self.assertFalse(out, 0)
 
     def test_30_nmap(self):
-
         #
         # Database tests
         #
@@ -2064,7 +2062,6 @@ def test_53_nmap_delete(self):
         self.assertEqual(count, 0)
 
     def test_40_passive(self):
-
         if DATABASE == "postgres":
             # FIXME: tests are broken with PostgreSQL & --no-bulk
             bulk_mode = random.choice(["--bulk", "--local-bulk"])
@@ -3304,7 +3301,6 @@ def test_54_passive_delete(self):
         self.assertEqual(count + new_count, total_count)
 
     def test_60_flow(self):
-
         # Unit tests #
 
         # Test _get_timeslots
@@ -4872,14 +4868,14 @@ def test_scans(self):
         self.assertEqual(res, 0)
         self.assertTrue(("  - locked (by %d)\n" % os.getpid()).encode() in out)
         # Attempt to lock it again
-        with (self.assertRaises(ivre.db.LockError)):
+        with self.assertRaises(ivre.db.LockError):
             ivre.db.db.agent.lock_scan(scanid)
         # Unlock it
         self.assertEqual(ivre.db.db.agent.unlock_scan(locked_scan), True)
         # Attempt to unlock it again
-        with (self.assertRaises(ivre.db.LockError)):
+        with self.assertRaises(ivre.db.LockError):
             ivre.db.db.agent.unlock_scan(locked_scan)
-        with (self.assertRaises(ivre.db.LockError)):
+        with self.assertRaises(ivre.db.LockError):
             ivre.db.db.agent.unlock_scan(ivre.db.db.agent.get_scan(scanid))
         # Check no scan is locked
         res, out, _ = RUN(["ivre", "runscansagentdb", "--list-scans"])
@@ -5049,7 +5045,6 @@ def _json_loads(data, deflt=None):
         shutil.rmtree("output")
 
     def test_50_view(self):
-
         #
         # Web server tests
         #