mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-10-01 00:02:34 +03:00
Merge pull request #2914 from markparonyan/mypy-utils-misc
fix(typing): resolve mypy errors in utils
This commit is contained in:
commit
ec4b04ba54
@ -48,7 +48,7 @@ def init_encryption(secrets_dir: str) -> None:
|
|||||||
os.makedirs(secrets_dir, exist_ok=True)
|
os.makedirs(secrets_dir, exist_ok=True)
|
||||||
with open(encryption_key_path, "w", encoding="utf-8") as f:
|
with open(encryption_key_path, "w", encoding="utf-8") as f:
|
||||||
# Use Fernet's base64-encoded key format
|
# Use Fernet's base64-encoded key format
|
||||||
f.write(key.decode() if isinstance(key, bytes) else key)
|
f.write(key.decode())
|
||||||
# Set restrictive permissions (owner read/write only)
|
# Set restrictive permissions (owner read/write only)
|
||||||
os.chmod(encryption_key_path, 0o600)
|
os.chmod(encryption_key_path, 0o600)
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
@ -58,7 +58,7 @@ def init_encryption(secrets_dir: str) -> None:
|
|||||||
try:
|
try:
|
||||||
with open(encryption_key_path, encoding="utf-8") as f:
|
with open(encryption_key_path, encoding="utf-8") as f:
|
||||||
key_content = f.read().strip()
|
key_content = f.read().strip()
|
||||||
key_bytes = key_content.encode() if isinstance(key_content, str) else key_content
|
key_bytes = key_content.encode()
|
||||||
_fernet = Fernet(key_bytes)
|
_fernet = Fernet(key_bytes)
|
||||||
log.debug("Encryption initialized successfully")
|
log.debug("Encryption initialized successfully")
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
|
|||||||
@ -126,7 +126,7 @@ def get_builtin_disks() -> List[str]:
|
|||||||
return builtin_disks
|
return builtin_disks
|
||||||
|
|
||||||
|
|
||||||
async def read_image_info(path: str, expected_image_type: str = None) -> dict:
|
async def read_image_info(path: str, expected_image_type: str | None = None) -> dict:
|
||||||
|
|
||||||
header_magic_len = 7
|
header_magic_len = 7
|
||||||
try:
|
try:
|
||||||
@ -155,7 +155,7 @@ async def read_image_info(path: str, expected_image_type: str = None) -> dict:
|
|||||||
return image_info
|
return image_info
|
||||||
|
|
||||||
|
|
||||||
async def discover_images(image_type: str, skip_image_paths: list = None) -> List[dict]:
|
async def discover_images(image_type: str, skip_image_paths: list | None = None) -> List[dict]:
|
||||||
"""
|
"""
|
||||||
Scan directories for available images
|
Scan directories for available images
|
||||||
"""
|
"""
|
||||||
@ -378,9 +378,9 @@ async def write_image(
|
|||||||
f"File '{image_path}' already exists, please choose a different name or remove the existing image"
|
f"File '{image_path}' already exists, please choose a different name or remove the existing image"
|
||||||
)
|
)
|
||||||
|
|
||||||
checksum = checksum.hexdigest()
|
checksum_hex = checksum.hexdigest()
|
||||||
image_dir = os.path.dirname(image_path)
|
image_dir = os.path.dirname(image_path)
|
||||||
duplicate_image = await images_repo.get_image_by_checksum(checksum, image_dir)
|
duplicate_image = await images_repo.get_image_by_checksum(checksum_hex, image_dir)
|
||||||
if duplicate_image:
|
if duplicate_image:
|
||||||
raise InvalidImageError(
|
raise InvalidImageError(
|
||||||
f"Image '{duplicate_image.filename}' with the same checksum already exists in '{image_dir}'"
|
f"Image '{duplicate_image.filename}' with the same checksum already exists in '{image_dir}'"
|
||||||
@ -396,5 +396,5 @@ async def write_image(
|
|||||||
log.warning(f"Could not remove '{tmp_path}'")
|
log.warning(f"Could not remove '{tmp_path}'")
|
||||||
|
|
||||||
return await images_repo.add_image(
|
return await images_repo.add_image(
|
||||||
image_name, image_type, image_size, image_path, checksum, checksum_algorithm="md5"
|
image_name, image_type, image_size, image_path, checksum_hex, checksum_algorithm="md5"
|
||||||
)
|
)
|
||||||
|
|||||||
@ -15,7 +15,7 @@ class FilterValidationError(Exception):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def validate_bpf_syntax(bpf_expression: str) -> Dict[str, Optional[str]]:
|
def validate_bpf_syntax(bpf_expression: str) -> Dict[str, Any]:
|
||||||
"""
|
"""
|
||||||
Validate BPF filter expression syntax using tcpdump.
|
Validate BPF filter expression syntax using tcpdump.
|
||||||
|
|
||||||
@ -78,7 +78,7 @@ def validate_filter_parameters(filter_type: str, values: List[Any]) -> None:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
# Define validation rules based on ubridge implementation
|
# Define validation rules based on ubridge implementation
|
||||||
VALIDATION_RULES = {
|
VALIDATION_RULES: Dict[str, Dict[str, Any]] = {
|
||||||
"frequency_drop": {
|
"frequency_drop": {
|
||||||
"params_count": 1,
|
"params_count": 1,
|
||||||
"ranges": [(-1, 32767)], # min, max
|
"ranges": [(-1, 32767)], # min, max
|
||||||
@ -173,11 +173,11 @@ def filter_inactive_filters(filters: Dict[str, List[Any]]) -> Dict[str, List[Any
|
|||||||
|
|
||||||
active_filters = {}
|
active_filters = {}
|
||||||
for filter_type, values in filters.items():
|
for filter_type, values in filters.items():
|
||||||
if not values or (isinstance(values, list) and len(values) == 0):
|
if not values:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# Normalize values (strip strings, convert to int)
|
# Normalize values (strip strings, convert to int)
|
||||||
normalized_values = []
|
normalized_values: List[Any] = []
|
||||||
for value in values:
|
for value in values:
|
||||||
if isinstance(value, str):
|
if isinstance(value, str):
|
||||||
normalized_values.append(value.strip("\n "))
|
normalized_values.append(value.strip("\n "))
|
||||||
@ -228,7 +228,7 @@ def validate_all_filters(filters: Dict[str, List[Any]]) -> None:
|
|||||||
return
|
return
|
||||||
|
|
||||||
for filter_type, values in filters.items():
|
for filter_type, values in filters.items():
|
||||||
if not values or (isinstance(values, list) and len(values) == 0):
|
if not values:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
validate_filter_parameters(filter_type, values)
|
validate_filter_parameters(filter_type, values)
|
||||||
|
|||||||
@ -372,10 +372,7 @@ module = [
|
|||||||
"gns3server.services.computes", # 4
|
"gns3server.services.computes", # 4
|
||||||
"gns3server.services.templates", # 18
|
"gns3server.services.templates", # 18
|
||||||
"gns3server.utils.asyncio.aiozipstream", # 8
|
"gns3server.utils.asyncio.aiozipstream", # 8
|
||||||
"gns3server.utils.encryption", # 2
|
|
||||||
"gns3server.utils.http_client", # 4
|
"gns3server.utils.http_client", # 4
|
||||||
"gns3server.utils.images", # 4
|
|
||||||
"gns3server.utils.packet_filter_validation", # 16
|
|
||||||
"gns3server.utils.websocket_to_websocket", # 12
|
"gns3server.utils.websocket_to_websocket", # 12
|
||||||
"gns3server.utils.zipfile_zstd._patcher", # 1
|
"gns3server.utils.zipfile_zstd._patcher", # 1
|
||||||
]
|
]
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user