Release 0.15.1

.devcontainer/api.dockerfile

@@ -1,11 +1,11 @@
 # pulls community scripts from git repo
-FROM python:3.10.6-slim AS GET_SCRIPTS_STAGE
+FROM python:3.10.8-slim AS GET_SCRIPTS_STAGE
 
 RUN apt-get update && \
     apt-get install -y --no-install-recommends git && \
     git clone https://github.com/amidaware/community-scripts.git /community-scripts
 
-FROM python:3.10.6-slim
+FROM python:3.10.8-slim
 
 ENV TACTICAL_DIR /opt/tactical
 ENV TACTICAL_READY_FILE ${TACTICAL_DIR}/tmp/tactical.ready

.github/workflows/ci-tests.yml

@@ -14,7 +14,7 @@ jobs:
     name: Tests
     strategy:
       matrix:
-        python-version: ["3.10.6"]
+        python-version: ["3.10.8"]
 
     steps:
       - uses: actions/checkout@v3
@@ -27,9 +27,10 @@ jobs:
           postgresql password: "pipeline123456"
 
       - name: Setup Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v3
+        uses: actions/setup-python@v4
         with:
           python-version: ${{ matrix.python-version }}
+          check-latest: true
 
       - name: Install redis
         run: |

.vscode/settings.json

@@ -1,7 +1,10 @@
 {
-  "python.defaultInterpreterPath": "api/tacticalrmm/env/bin/python",
+  "python.defaultInterpreterPath": "api/env/bin/python",
   "python.languageServer": "Pylance",
-  "python.analysis.extraPaths": ["api/tacticalrmm", "api/env"],
+  "python.analysis.extraPaths": [
+    "api/tacticalrmm",
+    "api/env"
+  ],
   "python.analysis.diagnosticSeverityOverrides": {
     "reportUnusedImport": "error",
     "reportDuplicateImport": "error",
@@ -22,7 +25,9 @@
     "**env/**"
   ],
   "python.formatting.provider": "black",
-  "mypy.targets": ["api/tacticalrmm"],
+  "mypy.targets": [
+    "api/tacticalrmm"
+  ],
   "mypy.runUsingActiveInterpreter": true,
   "editor.bracketPairColorization.enabled": true,
   "editor.guides.bracketPairs": true,
@@ -70,4 +75,4 @@
     "completeUnimported": true,
     "staticcheck": true
   }
-}
+}

ansible/roles/trmm_dev/defaults/main.yml

@@ -1,6 +1,6 @@
 ---
 user: "tactical"
-python_ver: "3.10.6"
+python_ver: "3.10.8"
 go_ver: "1.18.5"
 backend_repo: "https://github.com/amidaware/tacticalrmm.git"
 frontend_repo: "https://github.com/amidaware/tacticalrmm-web.git"

api/tacticalrmm/accounts/management/commands/reset_2fa.py

@@ -1,5 +1,6 @@
 import os
 import subprocess
+from contextlib import suppress
 
 import pyotp
 from django.core.management.base import BaseCommand
@@ -25,7 +26,7 @@ def handle(self, *args, **kwargs):
         nginx = "/etc/nginx/sites-available/frontend.conf"
         found = None
         if os.path.exists(nginx):
-            try:
+            with suppress(Exception):
                 with open(nginx, "r") as f:
                     for line in f:
                         if "server_name" in line:
@@ -35,8 +36,6 @@ def handle(self, *args, **kwargs):
                 if found:
                     rep = found.replace("server_name", "").replace(";", "")
                     domain = "".join(rep.split())
-            except:
-                pass
 
         code = pyotp.random_base32()
         user.totp_key = code

api/tacticalrmm/accounts/permissions.py

@@ -31,8 +31,8 @@ class RolesPerms(permissions.BasePermission):
     def has_permission(self, r, view) -> bool:
         if r.method == "GET":
             return _has_perm(r, "can_list_roles")
-        else:
-            return _has_perm(r, "can_manage_roles")
+
+        return _has_perm(r, "can_manage_roles")
 
 
 class APIKeyPerms(permissions.BasePermission):

api/tacticalrmm/agents/management/commands/bulk_delete_agents.py

@@ -64,7 +64,7 @@ def handle(self, *args, **kwargs):
                 try:
                     agent.delete()
                 except Exception as e:
-                    err = f"Failed to delete agent {agent.hostname}: {str(e)}"
+                    err = f"Failed to delete agent {agent.hostname}: {e}"
                     self.stdout.write(self.style.ERROR(err))
                 else:
                     deleted_count += 1

api/tacticalrmm/agents/models.py

@@ -1,6 +1,7 @@
 import asyncio
 import re
 from collections import Counter
+from contextlib import suppress
 from distutils.version import LooseVersion
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union, cast
 
@@ -130,8 +131,8 @@ def timezone(self) -> str:
         # return the default timezone unless the timezone is explicity set per agent
         if self.time_zone:
             return self.time_zone
-        else:
-            return get_core_settings().default_time_zone
+
+        return get_core_settings().default_time_zone
 
     @property
     def is_posix(self) -> bool:
@@ -232,12 +233,12 @@ def checks(self) -> Dict[str, Any]:
                 alert_severity = (
                     check.check_result.alert_severity
                     if check.check_type
-                    in [
+                    in (
                         CheckType.MEMORY,
                         CheckType.CPU_LOAD,
                         CheckType.DISK_SPACE,
                         CheckType.SCRIPT,
-                    ]
+                    )
                     else check.alert_severity
                 )
                 if alert_severity == AlertSeverity.ERROR:
@@ -333,8 +334,8 @@ def local_ips(self) -> str:
 
         if len(ret) == 1:
             return cast(str, ret[0])
-        else:
-            return ", ".join(ret) if ret else "error getting local ips"
+
+        return ", ".join(ret) if ret else "error getting local ips"
 
     @property
     def make_model(self) -> str:
@@ -344,7 +345,7 @@ def make_model(self) -> str:
             except:
                 return "error getting make/model"
 
-        try:
+        with suppress(Exception):
             comp_sys = self.wmi_detail["comp_sys"][0]
             comp_sys_prod = self.wmi_detail["comp_sys_prod"][0]
             make = [x["Vendor"] for x in comp_sys_prod if "Vendor" in x][0]
@@ -361,14 +362,10 @@ def make_model(self) -> str:
                     model = sysfam
 
             return f"{make} {model}"
-        except:
-            pass
 
-        try:
+        with suppress(Exception):
             comp_sys_prod = self.wmi_detail["comp_sys_prod"][0]
             return cast(str, [x["Version"] for x in comp_sys_prod if "Version" in x][0])
-        except:
-            pass
 
         return "unknown make/model"
 
@@ -479,7 +476,7 @@ def get_agent_policies(self) -> "Dict[str, Optional[Policy]]":
         models.prefetch_related_objects(
             [
                 policy
-                for policy in [self.policy, site_policy, client_policy, default_policy]
+                for policy in (self.policy, site_policy, client_policy, default_policy)
                 if policy
             ],
             "excluded_agents",
@@ -589,7 +586,7 @@ def run_script(
     def approve_updates(self) -> None:
         patch_policy = self.get_patch_policy()
 
-        severity_list = list()
+        severity_list = []
         if patch_policy.critical == "approve":
             severity_list.append("Critical")
 
@@ -621,17 +618,14 @@ def get_patch_policy(self) -> "WinUpdatePolicy":
         if not agent_policy:
             agent_policy = WinUpdatePolicy.objects.create(agent=self)
 
+        # Get the list of policies applied to the agent and select the
+        # highest priority one.
         policies = self.get_agent_policies()
 
-        processed_policies: List[int] = list()
         for _, policy in policies.items():
-            if (
-                policy
-                and policy.active
-                and policy.pk not in processed_policies
-                and policy.winupdatepolicy.exists()
-            ):
+            if policy and policy.active and policy.winupdatepolicy.exists():
                 patch_policy = policy.winupdatepolicy.first()
+                break
 
         # if policy still doesn't exist return the agent patch policy
         if not patch_policy:
@@ -683,7 +677,7 @@ def set_alert_template(self) -> "Optional[AlertTemplate]":
         policies = self.get_agent_policies()
 
         # loop through all policies applied to agent and return an alert_template if found
-        processed_policies: List[int] = list()
+        processed_policies: List[int] = []
         for key, policy in policies.items():
             # default alert_template will override a default policy with alert template applied
             if (
@@ -873,7 +867,7 @@ def serialize(agent: "Agent") -> Dict[str, Any]:
         return AgentAuditSerializer(agent).data
 
     def delete_superseded_updates(self) -> None:
-        try:
+        with suppress(Exception):
             pks = []  # list of pks to delete
             kbs = list(self.winupdates.values_list("kb", flat=True))
             d = Counter(kbs)
@@ -898,8 +892,6 @@ def delete_superseded_updates(self) -> None:
 
             pks = list(set(pks))
             self.winupdates.filter(pk__in=pks).delete()
-        except:
-            pass
 
     def should_create_alert(
         self, alert_template: "Optional[AlertTemplate]" = None
@@ -1018,16 +1010,16 @@ def value(self) -> Union[List[Any], bool, str]:
             return cast(List[str], self.multiple_value)
         elif self.field.type == CustomFieldType.CHECKBOX:
             return self.bool_value
-        else:
-            return cast(str, self.string_value)
+
+        return cast(str, self.string_value)
 
     def save_to_field(self, value: Union[List[Any], bool, str]) -> None:
-        if self.field.type in [
+        if self.field.type in (
             CustomFieldType.TEXT,
             CustomFieldType.NUMBER,
             CustomFieldType.SINGLE,
             CustomFieldType.DATETIME,
-        ]:
+        ):
             self.string_value = cast(str, value)
             self.save()
         elif self.field.type == CustomFieldType.MULTIPLE:

api/tacticalrmm/agents/permissions.py

@@ -122,5 +122,5 @@ def has_permission(self, r, view) -> bool:
             return _has_perm(r, "can_list_agent_history") and _has_perm_on_agent(
                 r.user, view.kwargs["agent_id"]
             )
-        else:
-            return _has_perm(r, "can_list_agent_history")
+
+        return _has_perm(r, "can_list_agent_history")

api/tacticalrmm/agents/serializers.py

@@ -100,21 +100,21 @@ def get_alert_template(self, obj):
 
         if not obj.alert_template:
             return None
-        else:
-            return {
-                "name": obj.alert_template.name,
-                "always_email": obj.alert_template.agent_always_email,
-                "always_text": obj.alert_template.agent_always_text,
-                "always_alert": obj.alert_template.agent_always_alert,
-            }
+
+        return {
+            "name": obj.alert_template.name,
+            "always_email": obj.alert_template.agent_always_email,
+            "always_text": obj.alert_template.agent_always_text,
+            "always_alert": obj.alert_template.agent_always_alert,
+        }
 
     def get_logged_username(self, obj) -> str:
         if obj.logged_in_username == "None" and obj.status == AGENT_STATUS_ONLINE:
             return obj.last_logged_in_user
         elif obj.logged_in_username != "None":
             return obj.logged_in_username
-        else:
-            return "-"
+
+        return "-"
 
     def get_italic(self, obj) -> bool:
         return obj.logged_in_username == "None" and obj.status == AGENT_STATUS_ONLINE

api/tacticalrmm/agents/utils.py

@@ -1,6 +1,7 @@
 import asyncio
 import tempfile
 import urllib.parse
+from pathlib import Path
 
 from django.conf import settings
 from django.http import FileResponse
@@ -54,9 +55,7 @@ def generate_linux_install(
         f"{core.mesh_site}/meshagents?id={mesh_id}&installflags=2&meshinstall={arch_id}"
     )
 
-    sh = settings.LINUX_AGENT_SCRIPT
-    with open(sh, "r") as f:
-        text = f.read()
+    text = Path(settings.LINUX_AGENT_SCRIPT).read_text()
 
     replace = {
         "agentDLChange": download_url,