Remove mypy and pytype in favour of pyrefly

.github/workflows/pytype.yml → .github/workflows/pyrefly.yml

@@ -1,18 +1,17 @@
-name: pytype checks
+name: Pyrefly type checking
 
 on:
   pull_request:
     types: [opened, synchronize, reopened]
 
 jobs:
-  pytype:
 
+  poetry:
     runs-on: ubuntu-latest
     strategy:
       matrix:
         os: [ubuntu-latest]
-        python-version: ["3.12"]  # Pytype won't suppoer higher than 3.12 - https://github.com/google/pytype/tree/main
-
+        python-version: ["3.12", "3.13", "3.14"]
     steps:
     - uses: actions/checkout@v2
     - name: Set up Python ${{ matrix.python-version }}
@@ -22,7 +21,7 @@ jobs:
     - name: Install dependencies
       run: |
         pip install poetry
-        poetry install
-    - name: Run pytype on dftimewolf and tests
+        poetry install --all-groups
+    - name: Pyrefly type checking
       run: |
-        poetry run pytype --config pytype.conf
+        poetry run pyrefly check

.pylintrc

@@ -432,7 +432,7 @@ indent-after-paren=4
 indent-string='  '
 
 # Maximum number of characters on a single line.
-max-line-length=140
+max-line-length=160
 
 # Maximum number of lines in a module.
 max-module-lines=1000

dftimewolf/cli/dftimewolf_recipes.py

@@ -85,7 +85,7 @@ def __init__(
     self._data_files_path = ''
     self._running_args: dict[str, typing.Any] = {}
     self._recipes_manager = recipes_manager.RecipesManager()
-    self._recipe: resources.Recipe = None  # type: ignore
+    self._recipe: resources.Recipe
     self._uuid = workflow_uuid or str(uuid.uuid4())
     logger.success(f'dfTimewolf tool initialized with UUID: {self._uuid}')
 
@@ -410,6 +410,8 @@ def SetupLogging(stdout_log: bool = False) -> None:
   file_handler.setLevel(logging.DEBUG)  # Always log DEBUG logs to files.
   logger.addHandler(file_handler)
 
+  assert file_handler.stream  # For typing
+
   if stdout_log:
     console_handler = logging.StreamHandler(stream=sys.stdout)
     colorize = sys.stdout.isatty() and not bool(os.environ.get('DFTIMEWOLF_NO_RAINBOW'))
@@ -419,9 +421,9 @@ def SetupLogging(stdout_log: bool = False) -> None:
     console_handler.setFormatter(logging_utils.WolfFormatter(
         handler_level=console_handler.level, colorize=colorize))
     logger.addHandler(console_handler)
-    logger.info(f'Logging to stdout and {file_handler.stream.name}')  # type: ignore
+    logger.info(f'Logging to stdout and {file_handler.stream.name}')
   else:
-    logger.info(f'Logging to {file_handler.stream.name}')  # type: ignore
+    logger.info(f'Logging to {file_handler.stream.name}')
 
 
 def RunTool() -> int:

dftimewolf/config.py

@@ -3,18 +3,18 @@
 
 import json
 
-from typing import Dict, Any
+from typing import Any
 
 from dftimewolf.lib import errors
 
 
 class Config(object):
   """Class that handles DFTimewolf's configuration parameters."""
 
-  _extra_config = {}  # type: Dict[str, Dict[str, Any]]
+  _extra_config: dict[str, dict[str, Any]] = {}
 
   @classmethod
-  def GetExtra(cls, name: str='') -> Dict[str, Any]:
+  def GetExtra(cls, name: str='') -> dict[str, Any]:
     """Retrieves extra configuration parameters.
 
     These parameters should be loaded through LoadExtra or LoadExtraData.

dftimewolf/lib/args_validator.py

@@ -10,7 +10,7 @@
 class AbstractValidator(abc.ABC):
   """Base class for validator objects."""
 
-  NAME: str = None  # type: ignore
+  NAME: str
 
   def __init__(self, dry_run: bool=False) -> None:
     """Initialize.

dftimewolf/lib/auth.py

@@ -1,16 +1,16 @@
 """Authentication module."""
 import os.path
-from typing import Optional
 
 import filelock
 from google.auth.transport.requests import Request
-from google.oauth2.credentials import Credentials
+from google.oauth2 import credentials as oauth_credentials
+from google.auth import external_account_authorized_user
 from google_auth_oauthlib.flow import InstalledAppFlow
 
 
 def GetGoogleOauth2Credential(
     scopes: list[str], credential_path: str, secret_path: str
-) -> Optional[Credentials]:
+) -> external_account_authorized_user.Credentials | oauth_credentials.Credentials:
   """Gets a Google Oauth2 credential.
 
   Args:
@@ -21,7 +21,7 @@ def GetGoogleOauth2Credential(
   Returns:
     Optional[Credentials]: Google Oauth2 credential.
   """
-  credentials: Optional[Credentials] = None
+  credentials: external_account_authorized_user.Credentials | oauth_credentials.Credentials | None = None
 
   # The credentials file stores the user's access and refresh tokens, and is
   # created automatically when the authorization flow completes for the first
@@ -30,13 +30,13 @@ def GetGoogleOauth2Credential(
   lock = filelock.FileLock(credentials_path + ".lock")  # pylint: disable=abstract-class-instantiated
   with lock:
     if os.path.exists(credentials_path):
-      credentials = Credentials.from_authorized_user_file(
-          credentials_path, scopes)  # type: ignore[no-untyped-call]
+      credentials = oauth_credentials.Credentials.from_authorized_user_file(
+          credentials_path, scopes)
 
     # If there are no (valid) credentials available, let the user log in.
     if not credentials or not credentials.valid:
       if credentials and credentials.expired and credentials.refresh_token:
-        credentials.refresh(Request())  # type: ignore
+        credentials.refresh(Request())
       else:
         secrets_path = os.path.join(os.path.expanduser("~"), secret_path)
         if not os.path.exists(secrets_path):

dftimewolf/lib/collectors/aws.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 """Creates an analysis VM and copies AWS volumes to it for analysis."""
 
-from typing import List, Optional, Callable
+from typing import Optional, Callable
 
 from libcloudforensics.providers.aws import forensics as aws_forensics
 from libcloudforensics.providers.aws.internal import account as aws_account
@@ -65,14 +65,14 @@ def __init__(self,
                      publish_message_callback=publish_message_callback)
     self.remote_profile_name = str()
     self.remote_zone = str()
-    self.source_account = None  # type: aws_account.AWSAccount
+    self.source_account: aws_account.AWSAccount
     self.incident_id = str()
-    self.remote_instance_id = None  # type: Optional[str]
-    self.volume_ids = []  # type: List[str]
+    self.remote_instance_id = str()
+    self.volume_ids: list[str] = []
     self.all_volumes = False
-    self.analysis_profile_name = None  # type: Optional[str]
-    self.analysis_zone = None  # type: Optional[str]
-    self.analysis_vm = None  # type: ec2.AWSInstance
+    self.analysis_profile_name = str()
+    self.analysis_zone = str()
+    self.analysis_vm: ec2.AWSInstance
     # See https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/device_naming.html
     self.device_suffixes = list('fghijklmnop')
 
@@ -86,23 +86,25 @@ def Process(self) -> None:
           volume_id=volume.volume_id,
           src_profile=self.remote_profile_name,
           dst_profile=self.analysis_profile_name)
-      self.analysis_vm.AttachVolume(
-          new_volume, self._FindNextAvailableDeviceName())
-      print('Volume {0:s} successfully copied to {1:s}'.format(
-          volume.volume_id, new_volume.volume_id))
 
-      container = containers.ForensicsVM(
-          name=self.analysis_vm.name,
-          evidence_disk=new_volume,
-          platform='aws')
-      self.StoreContainer(container)
+      if self.analysis_vm is not None:
+        self.analysis_vm.AttachVolume(
+            new_volume, self._FindNextAvailableDeviceName())
+        print('Volume {0:s} successfully copied to {1:s}'.format(
+            volume.volume_id, new_volume.volume_id))
+
+        container = containers.ForensicsVM(
+            name=str(self.analysis_vm.name),
+            evidence_disk=new_volume,
+            platform='aws')
+        self.StoreContainer(container)
 
   # pylint: disable=arguments-differ,too-many-arguments
   def SetUp(self,
             remote_profile_name: str,
             remote_zone: str,
             incident_id: str,
-            remote_instance_id: Optional[str]=None,
+            remote_instance_id: str,
             volume_ids: Optional[str]=None,
             all_volumes: bool=False,
             analysis_profile_name: Optional[str]=None,
@@ -193,7 +195,7 @@ def SetUp(self,
         dst_profile=self.analysis_profile_name,
     )
 
-  def _GetVolumesFromIds(self, volume_ids: List[str]) -> List[ebs.AWSVolume]:
+  def _GetVolumesFromIds(self, volume_ids: list[str]) -> list[ebs.AWSVolume]:
     """Gets volumes from an account by volume IDs.
 
     Args:
@@ -216,7 +218,7 @@ def _GetVolumesFromIds(self, volume_ids: List[str]) -> List[ebs.AWSVolume]:
 
   def _GetVolumesFromInstance(self,
                               instance_id: str,
-                              all_volumes: bool) -> List[ebs.AWSVolume]:
+                              all_volumes: bool) -> list[ebs.AWSVolume]:
     """Gets volumes to copy based on an instance name.
 
     Args:
@@ -237,7 +239,7 @@ def _GetVolumesFromInstance(self,
       return list(remote_instance.ListVolumes().values())
     return [remote_instance.GetBootVolume()]
 
-  def _FindVolumesToCopy(self) -> List[ebs.AWSVolume]:
+  def _FindVolumesToCopy(self) -> list[ebs.AWSVolume]:
     """Determines which volumes to copy depending on the collector's attributes.
 
     Returns:

dftimewolf/lib/collectors/aws_logging.py

@@ -4,7 +4,7 @@
 import json
 import tempfile
 import datetime
-from typing import Any, Dict, Optional, Callable
+from typing import Any, Optional, Callable
 
 from boto3 import session as boto3_session
 from botocore import exceptions as boto_exceptions
@@ -37,7 +37,7 @@ def __init__(self,
     self._query_filter: Optional[str] = None
     self._start_time: Optional[datetime.datetime] = None
     self._end_time: Optional[datetime.datetime] = None
-    self._region: str = None  # type: ignore
+    self._region = str()
 
   # pylint: disable=arguments-differ
   def SetUp(self,
@@ -90,7 +90,7 @@ def Process(self) -> None:
 
     cloudtrail_client = session.client('cloudtrail', region_name=self._region)
 
-    request_params: Dict[str, Any] = {}
+    request_params: dict[str, Any] = {}
     if self._query_filter:
       k, v = self._query_filter.split(',')
       filters = [{'AttributeKey': k, 'AttributeValue': v}]

dftimewolf/lib/collectors/aws_snapshot_s3_copy.py

@@ -2,7 +2,7 @@
 """Copies AWS EBS snapshots into AWS S3."""
 
 import time
-from typing import Any, Optional, Type, List, Callable
+from typing import Any, Optional, Type, Callable
 import boto3
 
 from libcloudforensics.providers.aws import forensics
@@ -64,7 +64,7 @@ def __init__(self,
     self.ec2: Any = None
     self.s3: Any = None
     self.iam_details: Any = None
-    self.aws_account: account.AWSAccount = None
+    self.aws_account: account.AWSAccount
     self.bucket_exists: bool = False
 
   # pylint: disable=arguments-differ
@@ -122,8 +122,7 @@ def PreProcess(self) -> None:
     if self.iam_details['profile']['created']:
       time.sleep(20) # Propagation delay
 
-  def Process(self, container: containers.AWSSnapshot
-              ) -> None:  # pytype: disable=signature-mismatch
+  def Process(self, container: containers.AWSSnapshot) -> None:  # pyrefly: ignore=[bad-override]
     """Perform the copy of the snapshot to S3."""
 
     # Aws accounts have thread safety issues. Create a unique one per thread
@@ -178,11 +177,11 @@ def _PickAvailabilityZone(self, subnet: Optional[str]='') -> str:
     # If we reached here, we have a problem
     raise AWSSnapshotS3CopyException('No suitable availability zone found')
 
-  def _CheckSnapshotsExist(self, snap_ids: List[str]) -> bool:
+  def _CheckSnapshotsExist(self, snap_ids: list[str]) -> bool:
     """Check the snapshots that we want to copy exist.
 
     Args:
-      snap_ids (List[str]): A list of snapshot IDs to look for.
+      snap_ids: A list of snapshot IDs to look for.
     Returns:
       True if the snapshots all exist and we have permissions to list them,
           False otherwise.