elpatron/octo-funnel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Activity

Update README

Browse Source
This commit is contained in:
Markus F.J. Busche
2025-09-20 14:59:24 +02:00
parent d379c8d2cd
commit 62ccfc4a00
6 changed files with 4 additions and 519 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -168,4 +168,5 @@ Thumbs.db
# Project specific # Project specific
OctoPrint_Tailscale_Funnel.egg-info/ OctoPrint_Tailscale_Funnel.egg-info/
**/OctoPrint_Tailscale_Funnel.egg-info/ **/OctoPrint_Tailscale_Funnel.egg-info/
dist/

52
README.md
View File

@@ -1,52 +0,0 @@
# OctoPrint Tailscale Funnel Plugin
This plugin makes your OctoPrint instance accessible from anywhere via Tailscale Funnel, without needing to configure port forwarding, dynamic DNS, or complex firewall settings.
## Features
* Enable/disable Tailscale Funnel access directly from OctoPrint's settings
* Monitor the current Funnel connection status
* Display the public URL for accessing OctoPrint remotely
* Configure the port to expose via Funnel
## Requirements
* OctoPrint 1.3.0 or higher
* Tailscale installed and configured on the system
* Python 3.7 or higher
## Installation
1. Install Tailscale on your system and ensure it's running (see https://tailscale.com/download/linux or run `curl -fsSL https://tailscale.com/install.sh | sh`)
2. Start Tailscale on your system (run `sudo tailscale up`)
3. Authenticate using the Tailscale URL (e.g. https://login.tailscale.com/a/<some random characters>)
4. Install the plugin through OctoPrint's plugin manager (go to Settings -> Plugins -> Install and search for "Tailscale Funnel")
5. Configure the plugin settings in OctoPrint's settings panel
6. Enable Funnel through the plugin interface
## Building from Source
If you want to build the plugin from source, please refer to the [BUILDING.md](octoprint_tailscale_funnel/BUILDING.md) file for detailed instructions.
## Configuration
The plugin adds a new section to OctoPrint's settings panel with the following options:
* **Port**: The port to expose via Funnel (default: 80)
* **Confirm Enable**: Require confirmation before enabling Funnel (default: True)
## Security Considerations
Enabling Funnel makes your OctoPrint instance accessible from the public internet. Only enable it when needed and disable it when finished. The plugin will show a confirmation dialog before enabling Funnel if the "Confirm Enable" option is checked.
## API Endpoints
The plugin exposes the following API endpoints:
* `GET /api/plugin/tailscale_funnel/status` - Get current Funnel status
* `POST /api/plugin/tailscale_funnel/enable` - Enable Tailscale Funnel
* `POST /api/plugin/tailscale_funnel/disable` - Disable Tailscale Funnel
## License
AGPLv3

1
README.md Symbolic link
View File

@@ -0,0 +1 @@
octoprint_tailscale_funnel/README.md

2
octoprint_tailscale_funnel/BUILDING.md
View File

@@ -55,7 +55,7 @@ This will create distribution files in the `dist/` directory:
First, install the zip utility if not already installed: First, install the zip utility if not already installed:
```bash ```bash
sudo apt install zip # On Ubuntu/Debian sudo apt install zip # On Ubuntu/Debian/Raspbian
``` ```
Then create a proper zip file: Then create a proper zip file:

230
octoprint_tailscale_funnel/__init__.py
View File

@@ -1,230 +0,0 @@
# coding=utf-8
from __future__ import absolute_import
import octoprint.plugin
from .tailscale_interface import TailscaleInterface, TailscaleError, TailscaleNotInstalledError, TailscaleNotRunningError
from .status_monitor import StatusMonitor
class TailscaleFunnelPlugin(octoprint.plugin.StartupPlugin,
octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.BlueprintPlugin):
def __init__(self):
self._logger = None
self.tailscale_interface = None
self.status_monitor = None
##~~ StartupPlugin mixin
def on_after_startup(self):
self._logger = self._plugin_manager.get_logger("octoprint_tailscale_funnel")
self._logger.info("Tailscale Funnel Plugin started")
self.tailscale_interface = TailscaleInterface(self._logger)
self.status_monitor = StatusMonitor(self)
self.status_monitor.start()
def on_shutdown(self):
if self.status_monitor:
self.status_monitor.stop()
##~~ SettingsPlugin mixin
def get_settings_defaults(self):
return dict(
enabled=False,
port=80,
confirm_enable=True
)
def get_settings_version(self):
return 1
def on_settings_migrate(self, target, current):
if current is None:
# Migrate from settings version 0 to 1
if self._settings.has(["funnel_enabled"]):
self._settings.set_boolean(["enabled"], self._settings.get_boolean(["funnel_enabled"]))
self._settings.remove(["funnel_enabled"])
def on_settings_save(self, data):
# Save the settings
octoprint.plugin.SettingsPlugin.on_settings_save(self, data)
# If enabled setting changed, update funnel status
if "enabled" in data:
enabled = self._settings.get_boolean(["enabled"])
port = self._settings.get_int(["port"])
try:
if enabled:
# Enable funnel
if self.tailscale_interface:
success = self.tailscale_interface.enable_funnel(port)
if success:
self._logger.info("Tailscale Funnel enabled on port {}".format(port))
else:
self._logger.error("Failed to enable Tailscale Funnel on port {}".format(port))
else:
# Disable funnel
if self.tailscale_interface:
success = self.tailscale_interface.disable_funnel(port)
if success:
self._logger.info("Tailscale Funnel disabled on port {}".format(port))
else:
self._logger.error("Failed to disable Tailscale Funnel on port {}".format(port))
except TailscaleError as e:
self._logger.error("Tailscale error: {}".format(str(e)))
except Exception as e:
self._logger.error("Unexpected error: {}".format(str(e)))
##~~ AssetPlugin mixin
def get_assets(self):
return dict(
js=["js/tailscale_funnel.js"],
css=["css/tailscale_funnel.css"],
less=["less/tailscale_funnel.less"]
)
##~~ TemplatePlugin mixin
def get_template_configs(self):
return [
dict(type="settings", custom_bindings=False)
]
##~~ BlueprintPlugin mixin
@octoprint.plugin.BlueprintPlugin.route("/status", methods=["GET"])
def get_status(self):
if not self.tailscale_interface:
return {"status": "error", "message": "Tailscale interface not initialized"}, 500
try:
# Check if tailscale is installed
if not self.tailscale_interface.is_tailscale_installed():
return {"status": "error", "message": "Tailscale is not installed"}, 500
# Check if tailscale is running
if not self.tailscale_interface.is_tailscale_running():
return {"status": "error", "message": "Tailscale is not running"}, 500
# Get funnel status
funnel_enabled = self.tailscale_interface.is_funnel_enabled()
public_url = self.tailscale_interface.get_public_url() if funnel_enabled else None
return {
"status": "success",
"data": {
"funnel_enabled": funnel_enabled,
"public_url": public_url,
"tailscale_installed": True,
"tailscale_running": True
}
}
except TailscaleNotInstalledError as e:
return {"status": "error", "message": "Tailscale is not installed: {}".format(str(e))}, 500
except TailscaleNotRunningError as e:
return {"status": "error", "message": "Tailscale is not running: {}".format(str(e))}, 500
except TailscaleError as e:
return {"status": "error", "message": "Tailscale error: {}".format(str(e))}, 500
except Exception as e:
self._logger.error("Unexpected error in get_status: {}".format(str(e)))
return {"status": "error", "message": "Unexpected error: {}".format(str(e))}, 500
@octoprint.plugin.BlueprintPlugin.route("/enable", methods=["POST"])
def enable_funnel(self):
if not self.tailscale_interface:
return {"status": "error", "message": "Tailscale interface not initialized"}, 500
try:
port = self._settings.get_int(["port"])
success = self.tailscale_interface.enable_funnel(port)
if success:
# Update settings
self._settings.set_boolean(["enabled"], True)
self._settings.save()
public_url = self.tailscale_interface.get_public_url()
return {
"status": "success",
"message": "Tailscale Funnel enabled successfully",
"data": {
"public_url": public_url
}
}
else:
return {"status": "error", "message": "Failed to enable Tailscale Funnel"}, 500
except TailscaleNotInstalledError as e:
return {"status": "error", "message": "Tailscale is not installed: {}".format(str(e))}, 500
except TailscaleError as e:
return {"status": "error", "message": "Tailscale error: {}".format(str(e))}, 500
except Exception as e:
self._logger.error("Unexpected error in enable_funnel: {}".format(str(e)))
return {"status": "error", "message": "Unexpected error: {}".format(str(e))}, 500
@octoprint.plugin.BlueprintPlugin.route("/disable", methods=["POST"])
def disable_funnel(self):
if not self.tailscale_interface:
return {"status": "error", "message": "Tailscale interface not initialized"}, 500
try:
port = self._settings.get_int(["port"])
success = self.tailscale_interface.disable_funnel(port)
if success:
# Update settings
self._settings.set_boolean(["enabled"], False)
self._settings.save()
return {
"status": "success",
"message": "Tailscale Funnel disabled successfully"
}
else:
return {"status": "error", "message": "Failed to disable Tailscale Funnel"}, 500
except TailscaleNotInstalledError as e:
return {"status": "error", "message": "Tailscale is not installed: {}".format(str(e))}, 500
except TailscaleError as e:
return {"status": "error", "message": "Tailscale error: {}".format(str(e))}, 500
except Exception as e:
self._logger.error("Unexpected error in disable_funnel: {}".format(str(e)))
return {"status": "error", "message": "Unexpected error: {}".format(str(e))}, 500
##~~ Softwareupdate hook
def get_update_information(self):
return dict(
tailscale_funnel=dict(
displayName="Tailscale Funnel Plugin",
displayVersion=self._plugin_version,
# version check: github repository
type="github_release",
user="markus",
repo="OctoPrint-Tailscale-Funnel",
current=self._plugin_version,
# update method: pip
pip="https://github.com/markus/OctoPrint-Tailscale-Funnel/archive/{target_version}.zip"
)
)
# If you want your plugin to be registered within OctoPrint under its own name,
# define it here.
__plugin_name__ = "Tailscale Funnel Plugin"
__plugin_pythoncompat__ = ">=3.7,<4" # python 3.7+
def __plugin_load__():
global __plugin_implementation__
__plugin_implementation__ = TailscaleFunnelPlugin()
global __plugin_hooks__
__plugin_hooks__ = {
"octoprint.plugin.softwareupdate.check_config": __plugin_implementation__.get_update_information
}

74
octoprint_tailscale_funnel/status_monitor.py
View File

@@ -1,74 +0,0 @@
# coding=utf-8
from __future__ import absolute_import
import threading
import time
class StatusMonitor:
def __init__(self, plugin, interval=30):
self.plugin = plugin
self.interval = interval
self._logger = plugin._logger
self._stop_event = threading.Event()
self._thread = None
def start(self):
"""
Start the status monitoring thread
"""
if self._thread is not None and self._thread.is_alive():
self._logger.warning("Status monitor is already running")
return
self._stop_event.clear()
self._thread = threading.Thread(target=self._monitor_loop)
self._thread.daemon = True
self._thread.start()
self._logger.info("Status monitor started")
def stop(self):
"""
Stop the status monitoring thread
"""
if self._thread is None:
return
self._stop_event.set()
self._thread.join()
self._thread = None
self._logger.info("Status monitor stopped")
def _monitor_loop(self):
"""
Main monitoring loop
"""
while not self._stop_event.is_set():
try:
# Check status if tailscale interface is available
if self.plugin.tailscale_interface:
# Get current funnel status
funnel_enabled = self.plugin.tailscale_interface.is_funnel_enabled()
# Update settings if needed
current_setting = self.plugin._settings.get_boolean(["enabled"])
if funnel_enabled != current_setting:
self.plugin._settings.set_boolean(["enabled"], funnel_enabled)
self.plugin._settings.save()
self._logger.info("Funnel status updated in settings: {}".format(funnel_enabled))
# Send a notification to the frontend
self.plugin._plugin_manager.send_plugin_message(
self.plugin._identifier,
{
"type": "funnel_status_change",
"enabled": funnel_enabled
}
)
except Exception as e:
self._logger.error("Error in status monitoring loop: {}".format(str(e)))
# Wait for the next interval or until stopped
if self._stop_event.wait(self.interval):
break

161
octoprint_tailscale_funnel/tailscale_interface.py
View File

@@ -1,161 +0,0 @@
# coding=utf-8
from __future__ import absolute_import
import subprocess
import json
import re
class TailscaleError(Exception):
"""Base exception for Tailscale interface errors"""
pass
class TailscaleNotInstalledError(TailscaleError):
"""Exception raised when Tailscale is not installed"""
pass
class TailscaleNotRunningError(TailscaleError):
"""Exception raised when Tailscale is not running"""
pass
class FunnelNotEnabledError(TailscaleError):
"""Exception raised when Funnel is not enabled"""
pass
class TailscaleInterface:
def __init__(self, logger=None):
self._logger = logger
def _run_command(self, command):
"""
Run a command and return the result
"""
try:
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
if result.returncode == 0:
return {"success": True, "output": result.stdout.strip(), "error": None}
else:
return {"success": False, "output": None, "error": result.stderr.strip()}
except subprocess.TimeoutExpired:
if self._logger:
self._logger.error("Command timed out: '{}'".format(command))
return {"success": False, "output": None, "error": "Command timed out"}
except Exception as e:
if self._logger:
self._logger.error("Error running command '{}': {}".format(command, str(e)))
return {"success": False, "output": None, "error": str(e)}
def is_tailscale_installed(self):
"""
Check if tailscale is installed
"""
result = self._run_command("which tailscale")
return result["success"] and result["output"] != ""
def is_tailscale_running(self):
"""
Check if tailscale is running
"""
result = self._run_command("tailscale status --json")
return result["success"]
def get_tailscale_status(self):
"""
Get tailscale status as JSON
"""
if not self.is_tailscale_installed():
raise TailscaleNotInstalledError("Tailscale is not installed")
result = self._run_command("tailscale status --json")
if result["success"]:
try:
return json.loads(result["output"])
except json.JSONDecodeError as e:
if self._logger:
self._logger.error("Error parsing tailscale status JSON: {}".format(str(e)))
raise TailscaleError("Error parsing tailscale status: {}".format(str(e)))
else:
raise TailscaleNotRunningError("Tailscale is not running: {}".format(result["error"]))
def is_funnel_enabled(self):
"""
Check if tailscale funnel is enabled
"""
if not self.is_tailscale_installed():
raise TailscaleNotInstalledError("Tailscale is not installed")
result = self._run_command("tailscale funnel status --json")
if result["success"]:
try:
status = json.loads(result["output"])
# Check if any funnel is active
for _, ports in status.get("AllowFunnel", {}).items():
if any(ports.values()):
return True
return False
except json.JSONDecodeError as e:
if self._logger:
self._logger.error("Error parsing funnel status JSON: {}".format(str(e)))
raise TailscaleError("Error parsing funnel status: {}".format(str(e)))
else:
# If the command fails, it might be because Funnel is not enabled
# Check if it's a specific error we can handle
if "funnel" in result["error"].lower() and "not enabled" in result["error"].lower():
return False
raise TailscaleError("Error checking funnel status: {}".format(result["error"]))
def enable_funnel(self, port):
"""
Enable tailscale funnel for the specified port
"""
if not self.is_tailscale_installed():
raise TailscaleNotInstalledError("Tailscale is not installed")
command = "tailscale funnel {} on".format(port)
result = self._run_command(command)
return result["success"]
def disable_funnel(self, port):
"""
Disable tailscale funnel for the specified port
"""
if not self.is_tailscale_installed():
raise TailscaleNotInstalledError("Tailscale is not installed")
command = "tailscale funnel {} off".format(port)
result = self._run_command(command)
return result["success"]
def get_public_url(self):
"""
Get the public URL for accessing the service via funnel
"""
if not self.is_tailscale_installed():
raise TailscaleNotInstalledError("Tailscale is not installed")
result = self._run_command("tailscale funnel status --json")
if result["success"]:
try:
status = json.loads(result["output"])
# Extract the public URL
# This is a simplified approach - in practice, you might need to parse more details
funnel_status = status.get("AllowFunnel", {})
if funnel_status:
# Get the first available funnel URL
for key, ports in funnel_status.items():
if any(ports.values()):
# Extract the domain from the key
match = re.search(r'https://([a-zA-Z0-9\-\.]+)', key)
if match:
return "https://{}".format(match.group(1))
return None
except (json.JSONDecodeError, KeyError) as e:
if self._logger:
self._logger.error("Error parsing funnel status for URL: {}".format(str(e)))
raise TailscaleError("Error parsing funnel status for URL: {}".format(str(e)))
else:
raise TailscaleError("Error getting funnel status: {}".format(result["error"]))