Improved State Management integrated with the UI upgrades, to evaluate failed zombie systemwide connections. And then prompt the user if they'd like to evaluate them for re-use, or close it.
This commit is contained in:
parent
fcace28890
commit
840c2ccb58
10 changed files with 376 additions and 34 deletions
|
|
@ -10,6 +10,7 @@ from core.utils.type_checker import type_checker
|
|||
|
||||
|
||||
from typing import Type, Optional
|
||||
import traceback
|
||||
|
||||
ERROR_TO_UI_MESSAGE = {
|
||||
ProfileActivationError: "The profile could not be enabled",
|
||||
|
|
@ -18,7 +19,7 @@ ERROR_TO_UI_MESSAGE = {
|
|||
ProfileModificationError: "WireGuard configuration could not be attached.",
|
||||
}
|
||||
|
||||
@type_checker
|
||||
|
||||
def translate_error(error: Type[Exception], profile_id: int) -> Result:
|
||||
"""Translate legacy core errors into actions and/or messages via Result Objects."""
|
||||
logger.error(str(error))
|
||||
|
|
@ -64,7 +65,22 @@ def translate_error(error: Type[Exception], profile_id: int) -> Result:
|
|||
return Result(valid=False, error_type=ResultError.MISSING_SOFTWARE, message=error_msg)
|
||||
|
||||
else:
|
||||
return Result(valid=False, error_type=ResultError.UNKNOWN, message="An unknown error occured.")
|
||||
# Extract the traceback from the exception object
|
||||
tb = error.__traceback__
|
||||
|
||||
# extract_tb returns a list of frames; the LAST one [-1] is where the error happened
|
||||
summary = traceback.extract_tb(tb)
|
||||
if summary:
|
||||
last_frame = summary[-1]
|
||||
func_name = last_frame.name # The function name
|
||||
line_no = last_frame.lineno # The line number
|
||||
file_name = last_frame.filename # The file path
|
||||
else:
|
||||
func_name, line_no, file_name = "Unknown", "Unknown", "Unknown"
|
||||
|
||||
error_msg = f"Error in {func_name} at {file_name}:{line_no} -> {str(error)}"
|
||||
logger.error(error_msg)
|
||||
return Result(valid=False, error_type=ResultError.UNKNOWN, message=error_msg)
|
||||
|
||||
|
||||
@type_checker
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
from core.models.system.SystemState import SystemState
|
||||
from core.models.session.SessionState import SessionState
|
||||
from core.models.enum_models import ProtocolChoice
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
|
@ -7,6 +8,7 @@ from typing import Optional
|
|||
@dataclass
|
||||
class GlobalState:
|
||||
systemwide_active: bool = False
|
||||
protocol: ProtocolChoice = None
|
||||
firewalled: bool = False
|
||||
dns_set: bool = False
|
||||
systemwide_profile_id: Optional[int] = None
|
||||
|
|
@ -37,6 +39,7 @@ class ProfileStateRepository:
|
|||
def turn_off_system_connection(self):
|
||||
"""Clear the system-wide memory cache."""
|
||||
self.global_state.systemwide_profile_id = None
|
||||
self.global_state.protocol = None
|
||||
self.global_state.firewalled = False
|
||||
self.global_state.dns_set = False
|
||||
self.global_state.systemwide_active = False
|
||||
|
|
@ -45,6 +48,7 @@ class ProfileStateRepository:
|
|||
def sync_system_state(self, state: SystemState):
|
||||
"""Called by SystemStateController to update memory cache."""
|
||||
self.global_state.systemwide_profile_id = state.profile_id
|
||||
self.global_state.protocol = state.protocol
|
||||
self.global_state.firewalled = state.firewalled
|
||||
self.global_state.dns_set = state.dns_set
|
||||
if state.profile_id:
|
||||
|
|
|
|||
|
|
@ -1,11 +1,36 @@
|
|||
from core.services.networking.general_connection_tools.connection_enable import establish_connection
|
||||
from core.controllers.ConfigurationController import ConfigurationController
|
||||
from core.infrastructure.global_state import ProfileStateRepository
|
||||
from core.controllers.ProfileController import ProfileController
|
||||
from core.controllers.SystemStateController import SystemStateController
|
||||
from core.controllers.SessionStateController import SessionStateController
|
||||
from core.models.system.SystemState import SystemState
|
||||
from core.models.enum_models import ProtocolChoice
|
||||
from core.models.Result import Result, ResultError
|
||||
from core.services.networking.systemwide import killswitch
|
||||
from core.services.networking.systemwide.general_tools import interface_tools
|
||||
from core.services.networking.systemwide.general_tools.test_any_protocol import test_ip_systemwide_for_any_protocol
|
||||
from core.services.networking.general_connection_tools.cross_protocol import kill_any_protocol
|
||||
from core.services.networking.systemwide.wireguard.wg_firewall_dns import revert_dns
|
||||
from core.services.networking.systemwide.dns_tools.process_dns_result import quick_domain_dns_check
|
||||
from core.observers.Notifications import Notifications, EventType
|
||||
from essentials.observers.ConnectionObserver import ConnectionObserver
|
||||
from core.Constants import Constants
|
||||
from core.errors.logger import logger
|
||||
|
||||
# generic
|
||||
from functools import partial
|
||||
|
||||
|
||||
SINGBOX_PROTOCOLS = (ProtocolChoice.HYSTERIA2, ProtocolChoice.VLESS)
|
||||
|
||||
def verify_systemwide_state(current_state: SystemState) -> bool:
|
||||
"""Confirm a valid connection, to update GlobalState (for UI)"""
|
||||
state_repo = ProfileStateRepository()
|
||||
state_repo.sync_system_state(current_state)
|
||||
state_repo.verified = True
|
||||
Notifications.emit(EventType.SYSTEM_ENABLE, current_state.profile_id)
|
||||
|
||||
def is_systemwide_on() -> bool:
|
||||
"""Returns a boolean if systemwide is connected. True is Yes, False is No"""
|
||||
state_repo = ProfileStateRepository()
|
||||
|
|
@ -36,6 +61,125 @@ def is_connected(profile_id: int) -> bool:
|
|||
return state_repo.is_profile_active(profile_id)
|
||||
|
||||
|
||||
def coordinate_evaluation(connection_observer: ConnectionObserver) -> Result:
|
||||
"""See if a connection can be salvaged. If not, rip it down & start over."""
|
||||
|
||||
# Claimed JSON state:
|
||||
if not SystemStateController.exists():
|
||||
error_msg = "State vanished inbetween the previous error prompt and now. That state is an external file, so that is technically possible, but a critical error."
|
||||
logger.error(error_msg)
|
||||
return Result(valid=False, error_type=ResultError.MISSING_DATA, message=error_msg)
|
||||
|
||||
json_state = SystemStateController.get()
|
||||
protocol = json_state.protocol
|
||||
profile_id = json_state.profile_id
|
||||
firewalled = json_state.firewalled
|
||||
|
||||
logger.info("Extracted JSON state data")
|
||||
|
||||
|
||||
# full profile:
|
||||
profile = ProfileController.get(profile_id)
|
||||
if not profile:
|
||||
error_msg = "Critical Failure where the profile to evaluate the connection can't be found. Resetting state."
|
||||
logger.error(error_msg)
|
||||
SystemStateController.dissolve()
|
||||
killed_firewall = killswitch.disarm()
|
||||
logger.info(f"killed_firewall: {killed_firewall.valid}")
|
||||
reverted_dns = revert_dns()
|
||||
logger.info(f"reverted_dns is {reverted_dns}")
|
||||
return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)
|
||||
logger.info("Got the relevant profile. Doing a pre-flight check")
|
||||
|
||||
#-----------------------
|
||||
# INTERFACE
|
||||
#-----------------------
|
||||
if protocol == ProtocolChoice.WIREGUARD:
|
||||
interface_up = interface_tools.exists("wg")
|
||||
elif protocol in SINGBOX_PROTOCOLS:
|
||||
interface_up = interface_tools.exists(Constants.SINGBOX_TUN_IF)
|
||||
else:
|
||||
interface_up = False
|
||||
logger.error("Unsupported protocol in systemwide state.")
|
||||
|
||||
# RULE OUT:
|
||||
if not interface_up:
|
||||
logger.info("Doing a new connection due to the interface being down")
|
||||
return establish_connection(profile, ignore={}, connection_observer=connection_observer)
|
||||
|
||||
#-----------------------
|
||||
# FIREWALL
|
||||
#-----------------------
|
||||
# Previously User requested Configuration requests
|
||||
user_wants_firewall = ConfigurationController.get_firewall_setting()
|
||||
|
||||
# Wireguard is voluntary opt-in, so they might not have the script installed,
|
||||
if not user_wants_firewall and protocol == ProtocolChoice.WIREGUARD:
|
||||
firewall_on = False
|
||||
# unless they explicitly opted out with WG, then check:
|
||||
else:
|
||||
if firewalled:
|
||||
firewall_on = killswitch.status()
|
||||
else:
|
||||
firewall_on = False
|
||||
|
||||
#-----------------------
|
||||
# # DNS
|
||||
#-----------------------
|
||||
if protocol == ProtocolChoice.WIREGUARD:
|
||||
target_interface = "wg"
|
||||
elif protocol in SINGBOX_PROTOCOLS:
|
||||
target_interface = Constants.SINGBOX_TUN_IF
|
||||
else:
|
||||
target_interface = "GARBAGE_FALSE"
|
||||
logger.error("Unsupported protocol in systemwide state.")
|
||||
|
||||
# this doesn't need sudo to run, and its just regular commands (no script setup):
|
||||
dns_on = quick_domain_dns_check(target_interface=target_interface)
|
||||
|
||||
logger.info("Evaluation Data Gathered")
|
||||
|
||||
# ------------------
|
||||
# DO API CALL & EVALUATE
|
||||
# ------------------
|
||||
test_result = test_ip_systemwide_for_any_protocol(profile_id=profile_id, protocol=protocol, connection_observer=connection_observer)
|
||||
|
||||
if not test_result.valid:
|
||||
logger.info("IP address check was not valid, so we're doing a new connection.")
|
||||
return establish_connection(profile, ignore={}, connection_observer=connection_observer)
|
||||
|
||||
if firewall_dns_in_line_with_expectations(firewalled=firewall_on, dns_set=dns_on, protocol=protocol):
|
||||
logger.info("IP address is connected and firewall & DNS expectations match reality. Update/Verify Global State that it's on")
|
||||
verify_systemwide_state(current_state=json_state) # this notifies the UI
|
||||
return Result(valid=True)
|
||||
|
||||
logger.error("Current connection did not meet the criteria to continue to use it as is.")
|
||||
return establish_connection(profile, ignore={}, connection_observer=connection_observer)
|
||||
|
||||
|
||||
def firewall_dns_in_line_with_expectations(firewalled: bool, dns_set: bool, protocol: ProtocolChoice) -> bool:
|
||||
"""We assume the IP address is okay. Is that enough?"""
|
||||
|
||||
if protocol != ProtocolChoice.WIREGUARD:
|
||||
if firewalled and dns_set:
|
||||
return True
|
||||
else:
|
||||
logger.info("Only wireguard is optional firewall/dns.")
|
||||
return False
|
||||
|
||||
# Previously User requested Configuration requests
|
||||
user_wants_firewall = ConfigurationController.get_firewall_setting()
|
||||
user_wants_dns = ConfigurationController.get_dns_setting()
|
||||
|
||||
if user_wants_firewall and not firewalled:
|
||||
return False
|
||||
|
||||
if user_wants_dns and not dns_set:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def evaluate_state_on_startup() -> Result:
|
||||
"""When the app first starts, evaluate pre-existing state"""
|
||||
|
||||
|
|
@ -44,9 +188,10 @@ def evaluate_state_on_startup() -> Result:
|
|||
#-----------------------
|
||||
systemwide_active = SystemStateController.get()
|
||||
if systemwide_active:
|
||||
error_msg = "A pre-existing systemwide state is active."
|
||||
logger.error(error_msg)
|
||||
return Result(valid=False, error_type=ResultError.STATE_CONFLICT, message=error_msg)
|
||||
error_msg = "The previous system-wide connection was not correctly turned off before the app was killed. Do you want to evaluate it? (no blindly kills it)"
|
||||
no_option = partial(kill_any_protocol, current_state=systemwide_active, active_protocol=systemwide_active.protocol)
|
||||
return Result(valid=False, no_option=no_option, error_type=ResultError.ZOMBIE_CONNECTION, message=error_msg)
|
||||
|
||||
|
||||
#-----------------------
|
||||
# SESSION
|
||||
|
|
@ -58,5 +203,3 @@ def evaluate_state_on_startup() -> Result:
|
|||
SessionStateController.remove(each_profile.id)
|
||||
|
||||
return Result(valid=True)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -126,6 +126,13 @@ class ThreadPlumbing:
|
|||
self.attempt_count += 1
|
||||
self._spawn_action(passed_result.function_choice)
|
||||
else:
|
||||
# if the result originally prescribed it's own "no option":
|
||||
no_option = passed_result.no_option
|
||||
if no_option:
|
||||
no_option()
|
||||
return
|
||||
|
||||
# otherwise do the original UI call:
|
||||
if self.ui_caller:
|
||||
self.ui_caller(passed_result)
|
||||
|
||||
|
|
@ -136,10 +143,12 @@ class ThreadPlumbing:
|
|||
|
||||
self.navigator.request_navigation.emit(auto_navigate_page)
|
||||
|
||||
|
||||
def normalize_seq(s):
|
||||
"""Ensures input is a tuple regardless of if it was a list or tuple."""
|
||||
return tuple(s) if s is not None else ()
|
||||
|
||||
|
||||
def resolve_prescription(result: Result, ENUM_MAP: dict, *args, **kwargs):
|
||||
"""
|
||||
Find the correct solution to the type of enum problem,
|
||||
|
|
@ -181,7 +190,30 @@ def resolve_prescription(result: Result, ENUM_MAP: dict, *args, **kwargs):
|
|||
if not function_requirements:
|
||||
return func
|
||||
|
||||
# Priority 3: Use original args/kwargs
|
||||
# Priority 3: Does it want the result itself?
|
||||
if 'result' in function_requirements:
|
||||
logger.info("result required; baking result into partial.")
|
||||
|
||||
# 1. Prepare the arguments we have available right now
|
||||
# Start with default_args from the ENUM_MAP if they exist
|
||||
current_args = {}
|
||||
if isinstance(default_args, dict):
|
||||
current_args.update(default_args)
|
||||
|
||||
# Inject the result object
|
||||
current_args['result'] = result
|
||||
|
||||
# 2. Use bind_partial so it doesn't fail if other args are still missing
|
||||
bound = sig.bind_partial(**current_args)
|
||||
|
||||
# 3. Fill in any function-defined defaults
|
||||
bound.apply_defaults()
|
||||
|
||||
# 4. Return the partial with the bound arguments
|
||||
return partial(func, **bound.arguments)
|
||||
|
||||
|
||||
# Priority 4: Use original args/kwargs
|
||||
try:
|
||||
bound = sig.bind(*args, **kwargs)
|
||||
bound.apply_defaults()
|
||||
|
|
@ -189,7 +221,7 @@ def resolve_prescription(result: Result, ENUM_MAP: dict, *args, **kwargs):
|
|||
except TypeError:
|
||||
pass # Not enough, move to priority 4
|
||||
|
||||
# Priority 4: Use original args/kwargs PLUS defaults
|
||||
# Priority 5: Use original args/kwargs PLUS defaults
|
||||
if default_args is None:
|
||||
return False
|
||||
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ class ResultError(Enum):
|
|||
EXPIRED_SUB = "expired_sub"
|
||||
ACTIVE_BILLING = "active_billing"
|
||||
STATE_CONFLICT = "state_conflict"
|
||||
ZOMBIE_CONNECTION = "zombie_connection"
|
||||
TICKET = "ticket"
|
||||
NO_TICKET = "no_ticket"
|
||||
RANDOM_TICKETS_OFF = "random_tickets_off"
|
||||
|
|
@ -60,6 +61,7 @@ class Result():
|
|||
message: Optional[str] = None
|
||||
goal: Optional[str] = None
|
||||
function_choice: Optional[partial] = None
|
||||
no_option: Optional[partial] = None
|
||||
navigate_to: Optional[str] = None
|
||||
|
||||
def user_message(self) -> str:
|
||||
|
|
|
|||
|
|
@ -70,3 +70,16 @@ class SystemState:
|
|||
@staticmethod
|
||||
def __get_state_path():
|
||||
return Constants.HV_STATE_HOME
|
||||
|
||||
|
||||
|
||||
# @dataclass
|
||||
# class StateEvaluation:
|
||||
# interface: bool
|
||||
# firewalled: bool
|
||||
# dns_set: bool
|
||||
# ip_address_correct: Optional[bool] = None
|
||||
# firewall_killed_internet: Optional[bool] = None
|
||||
# dns_killed_internet: Optional[bool] = None
|
||||
# has_internet: Optional[bool] = None
|
||||
# killed_vpn: Optional[bool] = None
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ def process_ALL_dns_results(target_interface: str) -> Result:
|
|||
|
||||
# ================= START LOOP =====================
|
||||
for each_command in list_of_commands:
|
||||
if each_command:
|
||||
each_SearchResult = process_ONE_dns_result(
|
||||
which_command=each_command,
|
||||
target_interface=target_interface,
|
||||
|
|
@ -93,6 +94,15 @@ def orchestrate_dns_check(target_interface: str) -> Result:
|
|||
return Result(valid=False, data=target_not_found)
|
||||
|
||||
|
||||
def quick_domain_dns_check(target_interface: str) -> bool:
|
||||
each_SearchResult = process_ONE_dns_result(
|
||||
which_command="domain",
|
||||
target_interface=target_interface,
|
||||
)
|
||||
if each_SearchResult.valid:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,64 @@
|
|||
|
||||
from core.services.networking.general_connection_tools.config_tools import extract_endpoint_ip
|
||||
from core.services.networking.general_connection_tools import ip
|
||||
from essentials.observers.ConnectionObserver import ConnectionObserver
|
||||
from core.models.enum_models import ProtocolChoice
|
||||
from core.utils.basic_operations.write_or_read_from_json import get_value_from_json_file
|
||||
from core.models.Result import Result, ResultError
|
||||
from core.errors.logger import logger
|
||||
from core.Constants import Constants
|
||||
|
||||
import os
|
||||
|
||||
|
||||
def get_singbox_ip(profile_id: int) -> str|None:
|
||||
"""Get the IP address for a Singbox based profile based on local disk."""
|
||||
config_path = f'{Constants.HV_SYSTEM_PROFILE_CONFIG_PATH}/{profile_id}/proxy.json'
|
||||
logger.info(f"config_path is {config_path}")
|
||||
|
||||
if not os.path.isfile(config_path):
|
||||
return None
|
||||
|
||||
server_ip = get_value_from_json_file(filepath=config_path, category="outbounds", key="server")
|
||||
return server_ip
|
||||
|
||||
|
||||
def test_ip_systemwide_for_any_protocol(profile_id: int, protocol: ProtocolChoice, connection_observer: ConnectionObserver) -> Result:
|
||||
"""Compare the real vs expected IP address."""
|
||||
|
||||
# protocol = profile.type
|
||||
# profile_id = profile.id
|
||||
|
||||
# what's expected:
|
||||
if protocol == ProtocolChoice.WIREGUARD:
|
||||
expected_ip_address = extract_endpoint_ip(profile_id)
|
||||
elif protocol in (ProtocolChoice.VLESS, ProtocolChoice.HYSTERIA2):
|
||||
expected_ip_address = get_singbox_ip(profile_id=profile_id)
|
||||
else:
|
||||
expected_ip_address = None
|
||||
|
||||
if not expected_ip_address:
|
||||
error_msg = "Could not get the expected IP address from local files."
|
||||
logger.error(error_msg)
|
||||
return Result(valid=False, error_type=ResultError.MISSING_DATA, message=error_msg)
|
||||
|
||||
# reality:
|
||||
ip_result = ip.test(connection_observer)
|
||||
|
||||
if not ip_result.valid:
|
||||
logger.error(ip_result.message)
|
||||
return ip_result # result object
|
||||
|
||||
real_ip = ip_result.data
|
||||
|
||||
print(f"Expected: {expected_ip_address} Reality: {real_ip}")
|
||||
|
||||
if real_ip == expected_ip_address:
|
||||
return Result(valid=True)
|
||||
else:
|
||||
error_msg = f"IP address does NOT match. Expected: {expected_ip_address} vs Reality: {real_ip}"
|
||||
return Result(valid=False, error_type=ResultError.LEAK_ISSUE, message=error_msg)
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
@ -212,7 +212,7 @@ def __establish_system_connection(
|
|||
current_state.dns_set = did_dns_work
|
||||
SystemStateController.update_or_create(current_state)
|
||||
|
||||
Notifications.emit(EventType.SYSTEM_ENABLE, profile_id)
|
||||
Notifications.emit(EventType.SYSTEM_ENABLE, profile.id)
|
||||
return Result(valid=True, data=current_state)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,49 +1,107 @@
|
|||
from core.models.Result import Result
|
||||
from core.models.Result import Result, ResultError
|
||||
from core.errors.logger import logger
|
||||
|
||||
|
||||
import inspect
|
||||
from functools import wraps
|
||||
from typing import Any
|
||||
from typing import get_origin, get_args, Union
|
||||
import types # Required for UnionType (the | operator)
|
||||
|
||||
def _get_type_error_details(check_type, value, origin, args_types):
|
||||
"""
|
||||
Helper to construct human-readable names for the expected and actual types.
|
||||
"""
|
||||
# Case 1: User used Type[T] but passed an instance of T
|
||||
if origin is type:
|
||||
inner_name = getattr(args_types[0], '__name__', 'the specified type')
|
||||
return f"a class/type (subclass of {inner_name})", f"an instance of {type(value).__name__}"
|
||||
|
||||
# Case 2: check_type is a tuple (common for Unions)
|
||||
if isinstance(check_type, tuple):
|
||||
type_names = ", ".join([getattr(t, '__name__', str(t)) for t in check_type])
|
||||
return f"one of ({type_names})", f"type {type(value).__name__}"
|
||||
|
||||
# Case 3: Standard single type
|
||||
type_name = getattr(check_type, '__name__', str(check_type))
|
||||
return type_name, f"type {type(value).__name__}"
|
||||
|
||||
def _validate_inner_types(value, origin, args_types, func, name):
|
||||
"""
|
||||
Helper to handle deep validation for generics like List[int] or Type[T].
|
||||
Returns (is_valid, error_message).
|
||||
"""
|
||||
if not args_types:
|
||||
return True, None
|
||||
|
||||
inner_type = args_types[0]
|
||||
|
||||
# Handle Type[T]: value must be a subclass of T
|
||||
if origin is type:
|
||||
if not issubclass(value, inner_type):
|
||||
inner_name = getattr(inner_type, '__name__', str(inner_type))
|
||||
return False, f"For function {func}, Argument '{name}' must be a subclass of {inner_name}, got {value.__name__}"
|
||||
return True, None
|
||||
|
||||
# Handle Collections: list, set, tuple
|
||||
if isinstance(value, (list, set, tuple)):
|
||||
for item in value:
|
||||
inner_origin = get_origin(inner_type)
|
||||
|
||||
# Handle Unions inside collections (e.g., List[Union[int, str]])
|
||||
if inner_origin is Union or (hasattr(types, "UnionType") and inner_origin is types.UnionType):
|
||||
if not isinstance(item, get_args(inner_type)):
|
||||
return False, f"For function {func}, Argument '{name}' contains element {type(item).__name__}, expected {inner_type}"
|
||||
# Handle standard types inside collections
|
||||
elif not isinstance(item, inner_type):
|
||||
inner_name = getattr(inner_type, '__name__', str(inner_type))
|
||||
return False, f"For function {func}, Argument '{name}' contains element {type(item).__name__}, expected {inner_name}"
|
||||
|
||||
return True, None
|
||||
|
||||
def type_checker(func):
|
||||
"""
|
||||
Type check wrapper.
|
||||
This type checks individual functions it's put on, and returns a Result object on failure.
|
||||
"""
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
# 1. Introspect the wrapped function
|
||||
sig = inspect.signature(func)
|
||||
|
||||
# 1. Bind arguments to names to handle both positional and keyword args
|
||||
try:
|
||||
# Bind the provided args/kwargs to the function's signature
|
||||
bound_args = sig.bind(*args, **kwargs)
|
||||
bound_args.apply_defaults()
|
||||
except TypeError as e:
|
||||
# This handles cases where the wrong number of arguments are passed
|
||||
error_msg = f"For {func}, {str(e)}"
|
||||
logger.error(error_msg)
|
||||
return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)
|
||||
|
||||
# 2. Perform type validation
|
||||
# 2. Validate each argument against its type hint
|
||||
for name, value in bound_args.arguments.items():
|
||||
expected_type = sig.parameters[name].annotation
|
||||
|
||||
# Skip parameters that don't have a type hint
|
||||
# Skip if no type hint is provided
|
||||
if expected_type is inspect.Parameter.empty:
|
||||
continue
|
||||
|
||||
if not isinstance(value, expected_type):
|
||||
error_msg = f"For function {func}, Argument '{name}' must be {expected_type.__name__}, got type {type(value).__name__}"
|
||||
origin = get_origin(expected_type)
|
||||
args_types = get_args(expected_type)
|
||||
|
||||
# Determine the type to use for isinstance()
|
||||
# Handles both typing.Union and the Python 3.10+ | operator (UnionType)
|
||||
is_union = origin is Union or (hasattr(types, "UnionType") and origin is types.UnionType)
|
||||
check_type = args_types if is_union else (origin if origin is not None else expected_type)
|
||||
|
||||
# Basic type validation
|
||||
if not isinstance(value, check_type):
|
||||
type_name, actual_name = _get_type_error_details(check_type, value, origin, args_types)
|
||||
error_msg = f"For function {func}, Argument '{name}' must be {type_name}, got {actual_name}"
|
||||
logger.error(error_msg)
|
||||
return Result(
|
||||
valid=False,
|
||||
error_type=ResultError.INVALID_INPUT,
|
||||
message=error_msg
|
||||
)
|
||||
return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)
|
||||
|
||||
# Deep validation for generics (e.g., List[int], Type[MyClass])
|
||||
# We only do this if it's not a Union, as Unions were handled by the basic isinstance check above
|
||||
if origin is not None and not is_union:
|
||||
is_valid, error_msg = _validate_inner_types(value, origin, args_types, func, name)
|
||||
if not is_valid:
|
||||
logger.error(error_msg)
|
||||
return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)
|
||||
|
||||
# 3. If all checks pass, call the original function
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
|
|
|||
Loading…
Reference in a new issue