Add missing type annotations

This commit also includes a return type change for util.get_validated_static_path

(backport to Yoda 1.9)

app.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-__copyright__ = 'Copyright (c) 2021-2023, Utrecht University'
+__copyright__ = 'Copyright (c) 2021-2024, Utrecht University'
 __license__   = 'GPLv3, see LICENSE'
 
 import threading
@@ -136,15 +136,16 @@ def static_loader() -> Optional[Response]:
 
     :returns: Static file
     """
-    result = get_validated_static_path(
+    static_dir, asset_name = get_validated_static_path(
         request.full_path,
         request.path,
         app.config.get('YODA_THEME_PATH'),
         app.config.get('YODA_THEME')
     )
-    if result is not None:
-        static_dir, asset_name = result
+    if static_dir and asset_name:
         return send_from_directory(static_dir, asset_name)
+    else:
+        return None
 
 
 @app.before_request

connman.py

@@ -14,20 +14,20 @@
 
 
 class Session(object):
-    def __init__(self, sid: int, irods: iRODSSession) -> None:
+    def __init__(self, sid: str, irods: iRODSSession) -> None:
         """Session object storing the iRODS session object.
 
         :param sid:   Flask session identifier
         :param irods: iRODS session
         """
-        self.sid: int             = sid               # Flask session identifier
+        self.sid: str             = sid               # Flask session identifier
         self.time: float          = time.time()       # Flask session start time
         self.irods: iRODSSession  = irods             # iRODS session
         self.irods_time: float    = time.time()       # iRODS session start time
         self.lock: threading.Lock = threading.Lock()
 
 
-sessions: Dict[int, Session] = dict()  # Custom session dict instead of Flask session (cannot pickle iRODS session)
+sessions: Dict[str, Session] = dict()  # Custom session dict instead of Flask session (cannot pickle iRODS session)
 lock: threading.Lock = threading.Lock()
 
 
@@ -54,7 +54,7 @@ def gc() -> None:
 gc_thread.start()
 
 
-def get(sid: int) -> Optional[iRODSSession]:
+def get(sid: str) -> Optional[iRODSSession]:
     """Retrieve iRODS session object from session."""
     if sid in sessions:
         s: Session = sessions[sid]
@@ -65,7 +65,7 @@ def get(sid: int) -> Optional[iRODSSession]:
     return s.irods
 
 
-def add(sid: int, irods: iRODSSession) -> None:
+def add(sid: str, irods: iRODSSession) -> None:
     """Add Flask sid and iRODS session to our custom session dict.
 
     :param sid:   Flask session identifier
@@ -80,7 +80,7 @@ def add(sid: int, irods: iRODSSession) -> None:
     print(f"[login]: Successfully connected to iRODS for session {sid}'")
 
 
-def release(sid: int) -> None:
+def release(sid: str) -> None:
     """Release a session.
 
     :param sid: Flask session identifier
@@ -93,7 +93,7 @@ def release(sid: int) -> None:
         s.lock.release()
 
 
-def clean(sid: int) -> None:
+def clean(sid: str) -> None:
     """Clean a session.
 
     :param sid: Flask session identifier
@@ -106,7 +106,7 @@ def clean(sid: int) -> None:
         print(f"[logout]: Cleanup session {sid}")
 
 
-def extend(sid: int) -> None:
+def extend(sid: str) -> None:
     """Extend session TTLs.
 
     :param sid: Flask session identifier

research/research.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+from __future__ import annotations
 
 __copyright__ = 'Copyright (c) 2021-2023, Utrecht University'
 __license__   = 'GPLv3, see LICENSE'
@@ -8,14 +9,15 @@
 import queue
 import threading
 import urllib.parse
-from typing import Iterator
+from typing import Iterator, Optional
 
 from flask import (
     abort, Blueprint, current_app as app, g, jsonify, make_response,
     render_template, request, Response, session, stream_with_context
 )
 from irods.data_object import iRODSDataObject
 from irods.exception import CAT_NO_ACCESS_PERMISSION, CAT_NO_ROWS_FOUND
+from irods.manager.data_object_manager import DataObjectManager
 from irods.message import iRODSMessage
 
 import api
@@ -29,24 +31,30 @@
 
 
 class Chunk:
-    def __init__(self, data_objects, path, number, size, data, resource):
-        self.data_objects = data_objects
-        self.path = path
-        self.number = number
-        self.size = size
-        self.data = data
-        self.resource = resource
+    def __init__(self,
+                 data_objects: Optional[DataObjectManager],
+                 path: Optional[str],
+                 number: int,
+                 size: int,
+                 data: Optional[str],
+                 resource: Optional[str]) -> None:
+        self.data_objects: Optional[DataObjectManager] = data_objects
+        self.path: Optional[str] = path
+        self.number: int = number
+        self.size: int = size
+        self.data: Optional[str] = data
+        self.resource: Optional[str] = resource
 
 
-q = queue.Queue(4)
-r = queue.Queue(1)
+q: queue.Queue[Chunk] = queue.Queue(4)
+r: queue.Queue[bool] = queue.Queue(1)
 
 
 def irods_writer() -> None:
     failure = False
     while True:
         chunk = q.get()
-        if chunk.path:
+        if chunk.path and chunk.data_objects is not None:
             if not failure:
                 try:
                     with chunk.data_objects.open(chunk.path, 'a', chunk.resource) as obj_desc:

setup.cfg

@@ -12,7 +12,7 @@ ignore_missing_imports = True
 warn_unreachable = True
 no_implicit_optional = True
 check_untyped_defs = False
-disallow_any_generics = True
+disallow_any_generics = False
 disallow_incomplete_defs = True
 disallow_untyped_calls = False
 disallow_untyped_defs = False

unit-tests/test_util.py

@@ -90,54 +90,61 @@ def test_static_loader_valid_path(self, mock_exists):
     def test_static_loader_invalid_path(self, mock_exists):
         mock_exists.side_effect = self.exists_return_value
         # Too short
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 "/?sawerw", "/", "/var/www/yoda/themes", "uu"
-            )
+            ),
+            ("", "")
         )
         # Path traversal attack
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 "/assets/../../../../etc/passwd?werwrwr",
                 "/assets/../../../../etc/passwd",
                 "/var/www/yoda/themes",
                 "uu",
-            )
+            ),
+            ("", "")
         )
         # non-printable characters
         full_path = "/assets/" + chr(13) + "img/logo.svg?werwer"
         path = "/assets/" + chr(13) + "img/logo.svg"
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 full_path, path, "/var/www/yoda/themes", "uu"
-            )
+            ),
+            ("", "")
         )
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 full_path, path, "/var/www/yoda/themes", "wur"
-            )
+            ),
+            ("", "")
         )
         # non-printable characters in asset name
         full_path = "/assets/img/l" + chr(13) + "ogo.svg?werwer"
         path = "/assets/img/l" + chr(13) + "ogo.svg"
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 full_path, path, "/var/www/yoda/themes", "uu"
-            )
+            ),
+            ("", "")
         )
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 full_path, path, "/var/www/yoda/themes", "wur"
-            )
+            ),
+            ("", "")
         )
         # .. in file name
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 "/assets/img/lo..go.svg?sklaerw",
                 "/assets/img/lo..go.svg?sklaerw",
                 "/var/www/yoda/themes",
                 "uu",
-            )
+            ),
+            ("", "")
         )
 
     @patch("os.path.exists")
@@ -171,22 +178,24 @@ def test_static_loader_module_valid_path(self, mock_exists):
     def test_static_loader_module_invalid_path(self, mock_exists):
         mock_exists.side_effect = self.exists_return_value
         # Invalid module name
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 "/../assets/../research/static/research/css/research.css?sklwrawe",
                 "/../assets/../research/static/research/css/research.css",
                 "/var/www/yoda/themes",
                 "uu",
-            )
+            ),
+            ("", "")
         )
         # Path traversal attack
-        self.assertIsNone(
+        self.assertEqual(
             get_validated_static_path(
                 "/group_manager/assets/../../../../../../etc/passwd?werwrwr",
                 "/group_manager/assets/../../../../../../etc/passwd",
                 "/var/www/yoda/themes",
                 "uu",
-            )
+            ),
+            ("", "")
         )
 
     def test_is_relative_url(self) -> None:

util.py

@@ -8,7 +8,7 @@
 import urllib
 from os import name, path
 from re import compile, fullmatch
-from typing import List, Optional, Tuple
+from typing import List, Tuple
 
 from werkzeug.security import safe_join
 from werkzeug.utils import secure_filename
@@ -90,8 +90,8 @@ def unicode_secure_filename(filename):
 
 
 def get_validated_static_path(
-    full_path, request_path, yoda_theme_path, yoda_theme
-) -> Optional[Tuple[str, str]]:
+    full_path: str, request_path: str, yoda_theme_path: str, yoda_theme: str
+) -> Tuple[str, str]:
     """
     Static files handling - recognisable through '/assets/'
     Confirms that input path is valid and return corresponding static path
@@ -113,13 +113,13 @@ def get_validated_static_path(
         _, asset_name = path.split(request_path)
         # Make sure asset_name is safe
         if asset_name != secure_filename(asset_name):
-            return
+            return "", ""
 
         if parts[0] == "assets":
             # Main assets
             static_dir = safe_join(user_static_area + "/static", *parts[1:])
             if not static_dir:
-                return
+                return "", ""
             user_static_filename = path.join(static_dir, asset_name)
             if not path.exists(user_static_filename):
                 static_dir = safe_join("/var/www/yoda/static", *parts[1:])
@@ -128,7 +128,7 @@ def get_validated_static_path(
             module = parts[0]
             # Make sure module name is safe
             if module != secure_filename(module):
-                return
+                return "", ""
 
             module_static_area = path.join(module, "static", module)
             user_module_static_filename = safe_join(
@@ -146,10 +146,13 @@ def get_validated_static_path(
             return "", ""
 
         full_path = path.join(static_dir, asset_name)
+
         # Check that path is correct
         if path.exists(full_path):
             return static_dir, asset_name
 
+    return "", ""
+
 
 def is_relative_url(url: str) -> bool:
     """