PORTify, a firewall passive port knocking by listening to Spotify tracks

[LavaTime]

No description provided.

[LavaTime]

@jules Could you perform a Code Review on the PR?

[LavaTime]

/gemini review

[LavaTime]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces an interesting project, PORTify, which uses Spotify's currently playing track to control firewall ports. The overall structure is good, leveraging Pydantic for settings and splitting logic into modules. However, I've identified several critical and high-severity issues that need to be addressed. The most critical issue is in libs/firewall.py, where the method of calling shell commands is unreliable and will likely fail on Windows. There are also several high-severity issues related to missing error handling and unsafe data access that could cause the application to crash unexpectedly. I've provided detailed comments and suggestions to fix these problems and improve the robustness and maintainability of the code.

[gemini-code-assist]

There are two critical issues with how subprocess.run is used throughout this file:

1. string.split() is not safe for parsing shell commands. It will fail for commands with quoted arguments containing spaces (like the PowerShell commands). You should use shlex.split() from the shlex module for robust parsing.
2. check=False with except CalledProcessError is incorrect. When check=False, subprocess.run will not raise a CalledProcessError on a non-zero exit code, making your except blocks unreachable. You must use check=True to catch command execution errors.

These issues make the firewall manipulation logic unreliable and prone to silent failures. This needs to be fixed for all subprocess.run calls in this file.

Suggested change

	command_to_run: List[str] = cls.LINUX_FIREWALL_RULE_TEMPLATE.substitute(
	maybe_delete=" delete", port_number=port
	).split()
	print(f"Closing port: {port}")
	try:
	subprocess.run(command_to_run, check=False, capture_output=True)
	except subprocess.CalledProcessError as e:
	print(f"Failed closing port: {port} Crashing aggresively")
	raise e
	command_to_run: List[str] = shlex.split(cls.LINUX_FIREWALL_RULE_TEMPLATE.substitute(
	maybe_delete=" delete", port_number=port
	))
	print(f"Closing port: {port}")
	try:
	subprocess.run(command_to_run, check=True, capture_output=True)
	except subprocess.CalledProcessError as e:
	print(f"Failed closing port: {port} Crashing aggressively")
	raise e

[gemini-code-assist]

The main loop does not handle exceptions that might be raised from get_current_playing_track_id (e.g., network issues, API changes). An unhandled exception here would crash the entire daemon. It's crucial to wrap the logic inside the while loop with a try...except block to gracefully handle errors and ensure the service remains running.

        try:
            current_playing_track_id: Optional[str] = get_current_playing_track_id(
                spotify_client=spotify_client
            )
            if current_playing_track_id == settings.SECRET_OPEN_TRACK_ID:
                print("User is playing the secret open track - Opening firewall's ports")
                open_firewall_ports(firewall=firewall, ports=settings.PORTS_TO_MANAGE)
                time_to_remove_rules = updated_closing_ports_countdown(
                    timeout=settings.OPEN_PORTS_TIMEOUT_MINUTES
                )
            elif current_playing_track_id == settings.SECRET_CLOSE_TRACK_ID:
                print("User is playing the secret close track - Closing firewall's ports")
                close_firewall_ports(firewall=firewall, ports=settings.PORTS_TO_MANAGE)
            else:
                maybe_close_ports_on_timeout(
                    firewall=firewall,
                    ports=settings.PORTS_TO_MANAGE,
                    time_to_remove_rules=time_to_remove_rules,
                )
        except Exception as e:
            # It's better to use a logger here
            print(f"An unexpected error occurred: {e}. Continuing...")

[gemini-code-assist]

This function is not safely handling the response from spotify_client.current_user_playing_track(). According to the spotipy documentation, this method can return None if no device is active or nothing is playing. Also, the item object can be None (e.g., during an ad) or may not have an id key. The current implementation will crash with a TypeError or KeyError in these common scenarios. You should add checks for None and use safe dictionary access with .get().

Suggested change

	user_current_playing_track: Dict[str, Any] = (
	spotify_client.current_user_playing_track()
	)
	is_playing: bool = user_current_playing_track["is_playing"]
	print(f"is_playing: {is_playing}")
	if not is_playing:
	return None
	current_playing_track_id: str = user_current_playing_track["item"]["id"]
	print(f"current_playing_track_id: {current_playing_track_id}")
	return current_playing_track_id
	user_current_playing_track: Optional[Dict[str, Any]] = (
	spotify_client.current_user_playing_track()
	)
	if not user_current_playing_track or not user_current_playing_track.get("is_playing"):
	return None
	item = user_current_playing_track.get("item")
	if not item:
	return None
	current_playing_track_id = item.get("id")
	if current_playing_track_id:
	print(f"current_playing_track_id: {current_playing_track_id}")
	return current_playing_track_id

[gemini-code-assist]

The cd PORTify command is likely incorrect. The python -m build command should be run from the project's root directory where pyproject.toml is located. There doesn't appear to be a subdirectory named PORTify that the user should change into. I suggest removing this line to avoid confusion.

[gemini-code-assist]

Throughout the application, print() is used for output. For a daemon/service, it's highly recommended to use a configurable logging framework (like Python's built-in logging module). This allows for better control over verbosity, log formatting, and directing output to files or system logs instead of just stdout. This applies to other print statements in this file as well (e.g., lines 66, 72).

[gemini-code-assist]

The pass statement here is redundant and can be removed.

[gemini-code-assist]

There is a typo in the rule comment: "automaticly" should be "automatically". This typo appears in all firewall rule templates in this file (lines 51, 54, 57). It's best to correct it everywhere for consistency.

Suggested change

	"sudo ufw$maybe_delete allow $port_number comment PORTify_automaticly_managed_rule:_allow_port_$port_number"
	"sudo ufw$maybe_delete allow $port_number comment PORTify_automatically_managed_rule:_allow_port_$port_number"

[gemini-code-assist]

Typo: "openinig" should be "opening".

Suggested change

	print("Rule exists, openinig port")
	print("Rule exists, opening port")

[gemini-code-assist]

Consider using a proper logging library like logging instead of print() statements. This provides more flexibility for log levels, formatting, and output destinations, which is crucial for a long-running service.

[gemini-code-assist]

While >3.8 is a valid version specifier, it's more conventional and clearer to use >=3.8 to indicate that Python 3.8 itself is a supported version, not just versions strictly greater than it.

Suggested change

	requires-python = ">3.8"
	requires-python = ">=3.8"