some linter and mypy fixes

dftimewolf/lib/collectors/gcp_logging.py

@@ -23,7 +23,7 @@
 
 def _CustomToAPIRepr(self: entries.ProtobufEntry) -> Dict[str, Any]:
   """API repr (JSON format) for entry."""
-  info = super(entries.ProtobufEntry, self).to_api_repr()  # type: ignore
+  info = super(entries.ProtobufEntry, self).to_api_repr()
   info['protoPayload'] = self.payload  # type: ignore
   return info  # type: ignore
 
@@ -62,9 +62,9 @@ def SetupLoggingClient(self) -> Any:
       logging.Client: A GCP logging client
     """
     if self._project_name:
-      return logging.Client(_use_grpc=False,  # type: ignore
+      return logging.Client(_use_grpc=False,
                                         project=self._project_name)
-    return logging.Client(_use_grpc=False)  # type: ignore
+    return logging.Client(_use_grpc=False)
 
   def ListPages(self, logging_client: Any) -> Any:
     """Returns pages based on a Cloud Logging filter

dftimewolf/lib/collectors/gsheets.py

@@ -39,7 +39,7 @@ def __init__(
     super(GoogleSheetsCollector, self).__init__(
         state, name=name, critical=critical)
     self._sheets_resource = None
-    self._credentials = None
+    self._credentials: Optional[Credentials] = None
     self._spreadsheet_id = ''
     self._sheet_names: List[str] = []
     # These are mandatory columns required by Timesketch.
@@ -127,13 +127,13 @@ def Process(self) -> None:
           '$ gcloud auth application-default login')
       self.ModuleError(str(exception), critical=True)
 
-  def _GetCredentials(self) -> Credentials:
+  def _GetCredentials(self) -> Optional[Credentials]:
     """Obtains API credentials for accessing the Google Sheets API.
 
     Returns:
       google.oauth2.credentials.Credentials: Google API credentials.
     """
-    credentials = None
+    credentials: Optional[Credentials] = None
 
     # The credentials file stores the user's access and refresh tokens, and is
     # created automatically when the authorization flow completes for the first
@@ -169,8 +169,9 @@ def _GetCredentials(self) -> Credentials:
           credentials = flow.run_console()
 
         # Save the credentials for the next run
-        with open(credentials_path, 'w') as token_file:
-          token_file.write(credentials.to_json())
+        if credentials:
+          with open(credentials_path, 'w') as token_file:
+            token_file.write(credentials.to_json())
 
     return credentials
 

dftimewolf/lib/collectors/workspace_audit.py

@@ -40,14 +40,15 @@ def __init__(self,
     """Initializes a Workspace Audit Log collector."""
     super(WorkspaceAuditCollector, self).__init__(state, name=name,
         critical=critical)
-    self._credentials = None
+    self._credentials: Optional[Credentials] = None
     self._application_name = ''
     self._filter_expression = ''
     self._user_key = 'all'
     self._start_time = None  # type: Optional[datetime.datetime]
     self._end_time = None # type: Optional[datetime.datetime]
 
-  def _BuildAuditResource(self, credentials: Credentials) -> discovery.Resource:
+  def _BuildAuditResource(self, credentials: Optional[Credentials]
+                          ) -> discovery.Resource:
     """Builds a reports resource object to use to request logs.
 
     Args:
@@ -59,13 +60,13 @@ def _BuildAuditResource(self, credentials: Credentials) -> discovery.Resource:
     service = discovery.build('admin', 'reports_v1', credentials=credentials)
     return service
 
-  def _GetCredentials(self) -> Credentials:
+  def _GetCredentials(self) -> Optional[Credentials]:
     """Obtains API credentials for accessing the Workspace audit API.
 
     Returns:
       google.oauth2.credentials.Credentials: Google API credentials.
     """
-    credentials = None
+    credentials: Optional[Credentials] = None
 
     # The credentials file stores the user's access and refresh tokens, and is
     # created automatically when the authorization flow completes for the first
@@ -103,8 +104,9 @@ def _GetCredentials(self) -> Credentials:
           credentials = flow.run_local_server()
 
         # Save the credentials for the next run
-        with open(credentials_path, 'w') as token_file:
-          token_file.write(credentials.to_json())
+        if credentials:
+          with open(credentials_path, 'w') as token_file:
+            token_file.write(credentials.to_json())
 
     return credentials
 

dftimewolf/lib/processors/turbinia_base.py

@@ -284,7 +284,7 @@ def GetCredentials(self, credentials_path: str,
                      client_secrets_path: str) -> Optional[Credentials]:
     """Authenticates the user using Google OAuth services."""
     scopes = ['openid', 'https://www.googleapis.com/auth/userinfo.email']
-    credentials = None
+    credentials: Optional[Credentials] = None
 
     # Load credentials file if it exists
     if os.path.exists(credentials_path):

dftimewolf/lib/processors/turbinia_gcp.py

@@ -142,6 +142,7 @@ def GetTelemetryEntry(self) -> Dict[str, str]:
     return telemetry_entry
 
   # pylint: disable=arguments-differ
+  # pylint: disable=too-many-arguments
   def SetUp(
       self,
       project: str,

mypy.ini

@@ -4,3 +4,4 @@ disable_error_code = override
 ignore_missing_imports = True
 strict = True
 exclude = _pb2.py|tests
+disallow_untyped_calls = False