lint: tidy up code some more, pylint is enforcing

.gitlab-ci.yml

@@ -51,7 +51,6 @@ test:pylint:
     - pip install -r requirements.txt --quiet
     - pip install pylint --quiet
     - pylint app
-  allow_failure: true
 
 pages:
   stage: deploy

.pylintrc

@@ -1,7 +1,9 @@
 [MASTER]
 disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
+extension-pkg-whitelist=pydantic
+fail-under=9.5
 ignored-classes=Column
-load-plugins=pylint_flask,pylint_flask_sqlalchemy
+load-plugins=pylint_flask,pylint_flask_sqlalchemy,pylint_pydantic
 max-line-length=120
 py-version=3.8
 suggestion-mode=yes

app/alarms.py

@@ -6,9 +6,9 @@ from app.extensions import db
 from app.models.alarms import Alarm
 
 
-def alarms_for(target: str) -> List[Alarm]:
+def alarms_for(target: BRN) -> List[Alarm]:
     return list(Alarm.query.filter(
-        Alarm.target == target
+        Alarm.target == str(target)
     ).all())
 
 

app/cli/db.py

@@ -50,7 +50,7 @@ def impot(model: db.Model) -> None:
                         line[idx] = datetime.datetime.strptime(line[idx], "%Y-%m-%d %H:%M:%S.%f")  # type: ignore
                 elif field_name in ["eotk", "auto_rotation", "smart"]:
                     # boolean fields
-                    line[idx] = line[idx] == "True"  # type: ignore
+                    line[idx] = line[idx] == "True"
                 elif field_name.endswith("_id") and line[idx] == "":
                     # integer foreign keys
                     line[idx] = None  # type: ignore

app/lists/bc2.py

@@ -1,3 +1,5 @@
+#  pylint: disable=too-few-public-methods
+
 import builtins
 from datetime import datetime
 from typing import List, Dict, Union, Any, Optional
@@ -33,14 +35,13 @@ def onion_alternative(origin: Origin) -> List[Dict[str, Any]]:
     url: Optional[str] = origin.onion()
     if url is None:
         return []
-    else:
+    return [{
-        return [{
+        "proto": "tor",
-            "proto": "tor",
+        "type": "eotk",
-            "type": "eotk",
+        "created_at": str(origin.added),
-            "created_at": str(origin.added),
+        "updated_at": str(origin.updated),
-            "updated_at": str(origin.updated),
+        "url": url}
-            "url": url}
+    ]
-        ]
 
 
 def proxy_alternative(proxy: Proxy) -> Dict[str, Any]:

app/lists/bridgelines.py

@@ -1,3 +1,5 @@
+#  pylint: disable=too-few-public-methods
+
 import builtins
 from typing import List, Iterable, Dict, Any, Optional
 
@@ -16,9 +18,12 @@ class Bridgelines(BaseModel):
     bridgelines: List[str] = Field(
         description="List of bridgelines, ready for use in a torrc file",
         examples=[
-            "Bridge obfs4 71.73.124.31:8887 E81B1237F6D13497B166060F55861565593CFF8E cert=b54NsV6tK1g+LHaThPOTCibdpx3wHm9NFe0PzGF1nwz+4M/tq6SkfOaShzPnZsIRCFRIHg iat-mode=0",
+            "Bridge obfs4 71.73.124.31:8887 E81B1237F6D13497B166060F55861565593CFF8E "
-            "Bridge obfs4 172.105.176.101:80 D18BC7E082D7EBF8E851029AC89A12A3F44A50BF cert=KHfAAUptXWRmLy3ehS9ETMO5luY06d0w7tEBDiAI0z62nC5Qo/APrzZxodkYWX2bNko/Mw iat-mode=0",
+            "cert=b54NsV6tK1g+LHaThPOTCibdpx3wHm9NFe0PzGF1nwz+4M/tq6SkfOaShzPnZsIRCFRIHg iat-mode=0",
-            "Bridge obfs4 141.101.36.55:9023 045EF272F08BC11CDB985889E4E9FE35DC6F9C67 cert=6KEdf/5aDSyuYEqvo14JE8Cks3i7PQtj9EFX2wTCiEaUPsp/I7eaOm4uSWdqwvV4vTVlFw iat-mode=0"
+            "Bridge obfs4 172.105.176.101:80 D18BC7E082D7EBF8E851029AC89A12A3F44A50BF "
+            "cert=KHfAAUptXWRmLy3ehS9ETMO5luY06d0w7tEBDiAI0z62nC5Qo/APrzZxodkYWX2bNko/Mw iat-mode=0",
+            "Bridge obfs4 141.101.36.55:9023 045EF272F08BC11CDB985889E4E9FE35DC6F9C67 "
+            "cert=6KEdf/5aDSyuYEqvo14JE8Cks3i7PQtj9EFX2wTCiEaUPsp/I7eaOm4uSWdqwvV4vTVlFw iat-mode=0 "
         ]
     )
 

app/lists/mirror_mapping.py

@@ -1,3 +1,5 @@
+#  pylint: disable=too-few-public-methods
+
 import builtins
 from typing import Dict, List, Union
 

app/models/__init__.py

@@ -24,7 +24,7 @@ class AbstractConfiguration(db.Model):  # type: ignore
 
     @property
     @abstractmethod
-    def brn(self) -> str:
+    def brn(self) -> BRN:
         raise NotImplementedError()
 
     def destroy(self) -> None:
@@ -80,12 +80,12 @@ class AbstractResource(db.Model):  # type: ignore
 
     def deprecate(self, *, reason: str) -> None:
         if self.deprecated is not None:
-            logging.info("Deprecating %s (reason=%s)", (self.brn, reason))
+            logging.info("Deprecating %s (reason=%s)", self.brn, reason)
             self.deprecated = datetime.utcnow()
             self.deprecation_reason = reason
             self.updated = datetime.utcnow()
         else:
-            logging.info("Not deprecating %s (reason=%s) because it's already deprecated", (self.brn, reason))
+            logging.info("Not deprecating %s (reason=%s) because it's already deprecated", self.brn, reason)
 
     def destroy(self) -> None:
         if self.deprecated is None:

app/models/activity.py

@@ -21,9 +21,9 @@ class Activity(db.Model):  # type: ignore
                  text: str,
                  added: Optional[datetime.datetime] = None,
                  **kwargs: Any) -> None:
-        if type(activity_type) != str or len(activity_type) > 20 or activity_type == "":
+        if not isinstance(activity_type, str) or len(activity_type) > 20 or activity_type == "":
             raise TypeError("expected string for activity type between 1 and 20 characters")
-        if type(text) != str:
+        if not isinstance(text, str):
             raise TypeError("expected string for text")
         super().__init__(id=id,
                          group_id=group_id,

app/models/automation.py

@@ -1,6 +1,7 @@
 import datetime
 import enum
 
+from app.brm.brn import BRN
 from app.extensions import db
 from app.models import AbstractConfiguration, AbstractResource
 
@@ -21,6 +22,16 @@ class Automation(AbstractConfiguration):
 
     logs = db.relationship("AutomationLogs", back_populates="automation")
 
+    @property
+    def brn(self) -> BRN:
+        return BRN(
+            group_id=0,
+            product="core",
+            provider="",
+            resource_type="automation",
+            resource_id=self.short_name
+        )
+
     def kick(self) -> None:
         self.enabled = True
         self.next_run = datetime.datetime.utcnow()
@@ -32,3 +43,13 @@ class AutomationLogs(AbstractResource):
     logs = db.Column(db.Text)
 
     automation = db.relationship("Automation", back_populates="logs")
+
+    @property
+    def brn(self) -> BRN:
+        return BRN(
+            group_id=0,
+            product="core",
+            provider="",
+            resource_type="automationlog",
+            resource_id=self.id
+        )

app/models/mirrors.py

@@ -1,6 +1,5 @@
 from typing import Optional, List
 
-from flask import current_app
 from tldextract import extract
 
 from app.brm.brn import BRN
@@ -20,8 +19,14 @@ class Origin(AbstractConfiguration):
     proxies = db.relationship("Proxy", back_populates="origin")
 
     @property
-    def brn(self) -> str:
+    def brn(self) -> BRN:
-        return f"brn:{current_app.config['GLOBAL_NAMESPACE']}:{self.group_id}:mirror:conf:origin/{self.domain_name}"
+        return BRN(
+            group_id=self.group_id,
+            product="mirror",
+            provider="conf",
+            resource_type="origin",
+            resource_id="self.domain_name"
+        )
 
     @classmethod
     def csv_header(cls) -> List[str]:

app/terraform/__init__.py

@@ -47,5 +47,5 @@ class BaseAutomation(metaclass=ABCMeta):
         :return: None
         """
         tmpl = jinja2.Template(template)
-        with open(self.working_directory(filename), 'w') as tf:
+        with open(self.working_directory(filename), 'w', encoding="utf-8") as tf:
             tf.write(tmpl.render(**kwargs))

app/terraform/alarms/proxy_cloudfront.py

@@ -6,6 +6,7 @@ from flask import current_app
 
 from app import app
 from app.alarms import get_or_create_alarm
+from app.brm.brn import BRN
 from app.extensions import db
 from app.models.mirrors import Proxy
 from app.models.alarms import AlarmState
@@ -14,7 +15,7 @@ from app.terraform import BaseAutomation
 
 def _cloudfront_quota() -> None:
     alarm = get_or_create_alarm(
-        f"brn:{current_app.config['GLOBAL_NAMESPACE']}:0:mirror:cloudfront:quota/distributions",
+        BRN.from_str(f"brn:{current_app.config['GLOBAL_NAMESPACE']}:0:mirror:cloudfront:quota/distributions"),
         "quota-usage"
     )
     alarm.last_updated = datetime.datetime.utcnow()

app/terraform/bridge/__init__.py

@@ -92,7 +92,7 @@ class BridgeAutomation(TerraformAutomation):
                 if len(parts) < 4:
                     continue
                 bridge = Bridge.query.filter(Bridge.id == output[len('bridge_bridgeline_'):]).first()
-                del(parts[3])
+                del parts[3]
                 bridge.bridgeline = " ".join(parts)
                 bridge.terraform_updated = datetime.datetime.utcnow()
         db.session.commit()

app/terraform/proxy/__init__.py

@@ -138,7 +138,6 @@ class ProxyAutomation(TerraformAutomation):
         self.create_missing_proxies()
         self.deprecate_orphaned_proxies()
         self.destroy_expired_proxies()
-        return None
 
     def tf_posthook(self, *, prehook_result: Any = None) -> None:
         self.import_state(self.tf_show())

app/terraform/terraform.py

@@ -44,12 +44,12 @@ class TerraformAutomation(BaseAutomation):
         :param full: include a Terraform refresh in the automation module run
         :return: success status and Terraform apply logs
         """
-        prehook_result = self.tf_prehook()
+        prehook_result = self.tf_prehook()  # pylint: disable=assignment-from-no-return
         self.tf_generate()
         self.tf_init()
         returncode, logs = self.tf_apply(refresh=self.always_refresh or full)
         self.tf_posthook(prehook_result=prehook_result)
-        return True if returncode == 0 else False, logs
+        return returncode == 0, logs
 
     def tf_apply(self, *,
                  refresh: bool = True,
@@ -60,7 +60,7 @@ class TerraformAutomation(BaseAutomation):
         # The following subprocess call takes external input, but is providing
         # the argument list as an array such that argument injection would be
         # ineffective.
-        tf = subprocess.run(  # nosec
+        tfcmd = subprocess.run(  # nosec
             ['terraform',
              'apply',
              '-auto-approve',
@@ -71,7 +71,7 @@ class TerraformAutomation(BaseAutomation):
              ],
             cwd=self.working_directory(),
             stdout=subprocess.PIPE)
-        return tf.returncode, tf.stdout.decode('utf-8')
+        return tfcmd.returncode, tfcmd.stdout.decode('utf-8')
 
     @abstractmethod
     def tf_generate(self) -> None:
@@ -92,11 +92,11 @@ class TerraformAutomation(BaseAutomation):
 
     def tf_output(self) -> Any:
         # The following subprocess call does not take any user input.
-        tf = subprocess.run(  # nosec
+        tfcmd = subprocess.run(  # nosec
             ['terraform', 'output', '-json'],
             cwd=self.working_directory(),
             stdout=subprocess.PIPE)
-        return json.loads(tf.stdout)
+        return json.loads(tfcmd.stdout)
 
     def tf_plan(self, *,
                 refresh: bool = True,
@@ -105,7 +105,7 @@ class TerraformAutomation(BaseAutomation):
         # The following subprocess call takes external input, but is providing
         # the argument list as an array such that argument injection would be
         # ineffective.
-        tf = subprocess.run(  # nosec
+        tfcmd = subprocess.run(  # nosec
             ['terraform',
              'plan',
              '-json',
@@ -114,7 +114,7 @@ class TerraformAutomation(BaseAutomation):
              f'-lock-timeout={str(lock_timeout)}m',
              ],
             cwd=self.working_directory())
-        return tf.returncode, tf.stdout.decode('utf-8')
+        return tfcmd.returncode, tfcmd.stdout.decode('utf-8')
 
     def tf_posthook(self, *, prehook_result: Any = None) -> None:
         """
@@ -126,7 +126,6 @@ class TerraformAutomation(BaseAutomation):
         :param prehook_result: the returned value of :func:`tf_prehook`
         :return: None
         """
-        pass
 
     def tf_prehook(self) -> Optional[Any]:
         """
@@ -138,7 +137,6 @@ class TerraformAutomation(BaseAutomation):
 
         :return: state that is useful to :func:`tf_posthook`, if required
         """
-        pass
 
     def tf_show(self) -> Any:
         # This subprocess call doesn't take any user input.

requirements.txt

@@ -12,6 +12,7 @@ jinja2~=3.0.2
 pydantic
 pylint-flask-sqlalchemy
 pylint-flask
+pylint-pydantic
 pyyaml~=6.0
 requests~=2.27.1
 sqlalchemy~=1.4.32