298 lines
12 KiB
Python
298 lines
12 KiB
Python
# coding=utf-8
|
|
from __future__ import absolute_import
|
|
|
|
import subprocess
|
|
import time
|
|
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 _run_first_success(self, commands):
|
|
"""
|
|
Try commands in order and return the first successful result.
|
|
"""
|
|
last_result = None
|
|
for command in commands:
|
|
result = self._run_command(command)
|
|
if self._logger:
|
|
self._logger.debug("tailscale cmd: '{}' -> rc_ok={}".format(command, result["success"]))
|
|
if result["success"]:
|
|
return result
|
|
# Try again with sudo -n if available/allowed, to avoid interactive prompt
|
|
sudo_command = "sudo -n " + command
|
|
sudo_result = self._run_command(sudo_command)
|
|
if self._logger:
|
|
self._logger.debug("tailscale cmd (sudo): '{}' -> rc_ok={}".format(sudo_command, sudo_result["success"]))
|
|
if sudo_result["success"]:
|
|
return sudo_result
|
|
last_result = result
|
|
return last_result if last_result is not None else {"success": False, "output": None, "error": "No commands executed"}
|
|
|
|
def is_tailscale_installed(self):
|
|
"""
|
|
Check if tailscale is installed
|
|
"""
|
|
# Try PATH lookup, absolute path, or a sudo check that doesn't prompt
|
|
candidates = [
|
|
"command -v tailscale",
|
|
"test -x /usr/bin/tailscale && echo /usr/bin/tailscale",
|
|
"test -x /usr/local/bin/tailscale && echo /usr/local/bin/tailscale",
|
|
]
|
|
result = self._run_first_success(candidates)
|
|
if result and result["success"] and result.get("output"):
|
|
return True
|
|
# Fallback: ask tailscale for version via sudo -n (non-interactive)
|
|
version_check = self._run_command("sudo -n tailscale version")
|
|
return version_check["success"]
|
|
|
|
def is_tailscale_running(self):
|
|
"""
|
|
Check if tailscale is running
|
|
"""
|
|
# Prefer non-interactive sudo to avoid PATH/permission issues
|
|
result = self._run_first_success([
|
|
"sudo -n tailscale status --json",
|
|
"tailscale status --json",
|
|
"/usr/bin/tailscale status --json",
|
|
"/usr/local/bin/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_first_success([
|
|
"sudo -n tailscale funnel status --json",
|
|
"tailscale funnel status --json",
|
|
"/usr/bin/tailscale funnel status --json",
|
|
"/usr/local/bin/tailscale funnel status --json",
|
|
])
|
|
if result["success"]:
|
|
try:
|
|
status = json.loads(result["output"]) if result["output"] else {}
|
|
|
|
# New CLI JSON structure (since ~1.44+): has keys like "TCP" and "Web"
|
|
tcp_cfg = status.get("TCP", {})
|
|
if isinstance(tcp_cfg, dict) and tcp_cfg:
|
|
for _port, cfg in tcp_cfg.items():
|
|
# cfg may contain flags like {"HTTP": true} or {"HTTPS": true}
|
|
if isinstance(cfg, dict) and any(bool(v) for v in cfg.values()):
|
|
return True
|
|
|
|
# Old CLI JSON structure: {"AllowFunnel": {"https://<domain>": {"80": true}}}
|
|
allow = status.get("AllowFunnel", {})
|
|
if isinstance(allow, dict) and allow:
|
|
for _key, ports in allow.items():
|
|
if isinstance(ports, dict) and any(bool(v) for v in 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")
|
|
|
|
# If already enabled, consider success
|
|
try:
|
|
if self.is_funnel_enabled():
|
|
return True
|
|
except Exception:
|
|
# ignore and continue to try enabling
|
|
pass
|
|
|
|
# Since newer Tailscale versions changed the CLI, try new syntax first,
|
|
# then fall back to the legacy commands.
|
|
# 1) Ensure a serve mapping exists
|
|
serve_cmd_new = "tailscale serve --bg --http {p} --set-path {p} http://127.0.0.1:{p}".format(p=port)
|
|
serve_cmd_old = "tailscale serve --bg https http://127.0.0.1:{p}".format(p=port)
|
|
self._run_first_success([serve_cmd_new, serve_cmd_old])
|
|
|
|
# 2) Enable funnel
|
|
funnel_cmd_new = "tailscale funnel --bg {p}".format(p=port)
|
|
funnel_cmd_old = "tailscale funnel {p} on".format(p=port)
|
|
result = self._run_first_success([funnel_cmd_new, funnel_cmd_old])
|
|
|
|
# Validate by querying status
|
|
if result["success"] and self.is_funnel_enabled():
|
|
return True
|
|
return False
|
|
|
|
def disable_funnel(self, port):
|
|
"""
|
|
Disable tailscale funnel for the specified port
|
|
"""
|
|
if not self.is_tailscale_installed():
|
|
raise TailscaleNotInstalledError("Tailscale is not installed")
|
|
|
|
# Execute multiple disable attempts tolerantly (idempotent)
|
|
commands = [
|
|
# Try to disable funnel first (new + legacy)
|
|
"tailscale funnel reset",
|
|
"tailscale funnel {p} off".format(p=port),
|
|
# Then disable serve mappings (newer variants)
|
|
"tailscale serve --http={p} off".format(p=port),
|
|
"tailscale serve --https={p} off".format(p=port),
|
|
]
|
|
|
|
any_succeeded = False
|
|
for cmd in commands:
|
|
res = self._run_first_success([cmd])
|
|
if res and res.get("success"):
|
|
any_succeeded = True
|
|
else:
|
|
# Treat "not enabled"/"no such" errors as harmless for idempotency
|
|
err = (res or {}).get("error") or ""
|
|
if isinstance(err, str) and (
|
|
"not enabled" in err.lower()
|
|
or "no handlers" in err.lower()
|
|
or "no such" in err.lower()
|
|
or "not found" in err.lower()
|
|
or "already off" in err.lower()
|
|
):
|
|
any_succeeded = True
|
|
|
|
# Poll for a short time until status reflects disabled
|
|
for _ in range(6): # ~3.0s total
|
|
try:
|
|
if not self.is_funnel_enabled():
|
|
return True
|
|
except Exception:
|
|
# If status check fails, assume disabled if we attempted commands
|
|
if any_succeeded:
|
|
return True
|
|
time.sleep(0.5)
|
|
|
|
# Final status check
|
|
try:
|
|
return not self.is_funnel_enabled()
|
|
except Exception:
|
|
return any_succeeded
|
|
|
|
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_first_success([
|
|
"sudo -n tailscale funnel status --json",
|
|
"tailscale funnel status --json",
|
|
"/usr/bin/tailscale funnel status --json",
|
|
"/usr/local/bin/tailscale funnel status --json",
|
|
])
|
|
if result["success"]:
|
|
try:
|
|
status = json.loads(result["output"]) if result["output"] else {}
|
|
|
|
# New JSON shape
|
|
web = status.get("Web", {})
|
|
if isinstance(web, dict) and web:
|
|
for hostport, mapping in web.items():
|
|
# hostport like "octopi.warbler-bearded.ts.net:80"
|
|
domain = hostport.split(":")[0]
|
|
# If handler paths exist, pick the first path
|
|
path = "/"
|
|
handlers = mapping.get("Handlers", {}) if isinstance(mapping, dict) else {}
|
|
if handlers:
|
|
first_path = next(iter(handlers.keys()))
|
|
if isinstance(first_path, str):
|
|
path = first_path if first_path.startswith("/") else "/" + first_path
|
|
scheme = "https" if hostport.endswith(":443") else "http"
|
|
return "{}://{}{}".format(scheme, domain, path)
|
|
|
|
# Legacy JSON shape
|
|
allow = status.get("AllowFunnel", {})
|
|
if isinstance(allow, dict) and allow:
|
|
for key, ports in allow.items():
|
|
if any(ports.values()):
|
|
match = re.search(r'https?://([a-zA-Z0-9\-\.]+)', key)
|
|
if match:
|
|
# Assume https if legacy provided https, else http
|
|
scheme = "https" if key.startswith("https://") else "http"
|
|
return "{}://{}".format(scheme, 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"])) |