/
/
/
1"""Helpers for validating redirect URLs in OAuth/auth flows."""
2
3from __future__ import annotations
4
5import ipaddress
6import logging
7from typing import TYPE_CHECKING
8from urllib.parse import quote, urlparse
9
10from music_assistant.constants import MASS_LOGGER_NAME
11
12if TYPE_CHECKING:
13 from aiohttp import web
14
15LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.redirect_validation")
16
17# Allowed redirect URI patterns
18# Add custom URL schemes for mobile apps here
19ALLOWED_REDIRECT_PATTERNS = [
20 # Custom URL schemes for mobile apps
21 "musicassistant://", # Music Assistant mobile app
22 # Home Assistant domains
23 "https://my.home-assistant.io/",
24 "http://homeassistant.local/",
25 "https://homeassistant.local/",
26]
27
28
29def is_allowed_redirect_url(
30 url: str,
31 request: web.Request | None = None,
32 base_url: str | None = None,
33) -> tuple[bool, str]:
34 """
35 Validate if a redirect URL is allowed for OAuth/auth flows.
36
37 Security rules (in order of priority):
38 1. Must use http, https, or registered custom scheme (e.g., musicassistant://)
39 2. Same origin as the request - auto-allowed (trusted)
40 3. Localhost (127.0.0.1, ::1, localhost) - auto-allowed (trusted)
41 4. Private network IPs (RFC 1918) - auto-allowed (trusted)
42 5. Configured base_url - auto-allowed (trusted)
43 6. Matches allowed redirect patterns - auto-allowed (trusted)
44 7. Everything else - requires user consent (external)
45
46 :param url: The redirect URL to validate.
47 :param request: Optional aiohttp request to compare origin.
48 :param base_url: Optional configured base URL to allow.
49 :return: Tuple of (is_valid, category) where category is:
50 - "trusted": Auto-allowed, no consent needed
51 - "external": Valid but requires user consent
52 - "blocked": Invalid/dangerous URL
53 """
54 if not url:
55 return False, "blocked"
56
57 try:
58 parsed = urlparse(url)
59
60 # Check for custom URL schemes (mobile apps)
61 for pattern in ALLOWED_REDIRECT_PATTERNS:
62 if url.startswith(pattern):
63 LOGGER.debug("Redirect URL trusted (pattern match): %s", url)
64 return True, "trusted"
65
66 # Only http/https for web URLs
67 if parsed.scheme not in ("http", "https"):
68 LOGGER.warning("Redirect URL blocked (invalid scheme): %s", url)
69 return False, "blocked"
70
71 hostname = parsed.hostname
72 if not hostname:
73 LOGGER.warning("Redirect URL blocked (no hostname): %s", url)
74 return False, "blocked"
75
76 # 1. Same origin as request - always trusted
77 if request:
78 request_host = request.host
79 if parsed.netloc == request_host:
80 LOGGER.debug("Redirect URL trusted (same origin): %s", url)
81 return True, "trusted"
82
83 # 2. Localhost - always trusted (for development and mobile app testing)
84 if hostname in ("localhost", "127.0.0.1", "::1"):
85 LOGGER.debug("Redirect URL trusted (localhost): %s", url)
86 return True, "trusted"
87
88 # 3. Private network IPs - always trusted (for local network access)
89 if _is_private_ip(hostname):
90 LOGGER.debug("Redirect URL trusted (private IP): %s", url)
91 return True, "trusted"
92
93 # 4. Configured base_url - always trusted
94 if base_url:
95 base_parsed = urlparse(base_url)
96 if parsed.netloc == base_parsed.netloc:
97 LOGGER.debug("Redirect URL trusted (base_url): %s", url)
98 return True, "trusted"
99
100 # If we get here, URL is external and requires user consent
101 LOGGER.info("Redirect URL is external (requires consent): %s", url)
102 return True, "external"
103
104 except Exception:
105 LOGGER.exception("Error validating redirect URL")
106 return False, "blocked"
107
108
109def build_code_redirect_url(
110 return_url: str,
111 token: str,
112 extra_params: dict[str, str] | None = None,
113) -> str:
114 """
115 Build a redirect URL with the auth code appended to the query string.
116
117 The code (and any extra params) are inserted before a possible hash fragment
118 so they end up in the query string rather than inside the fragment.
119
120 :param return_url: The validated destination URL to redirect to.
121 :param token: The auth token to pass along as the ``code`` query parameter.
122 :param extra_params: Optional additional query parameters to append.
123 """
124 params = f"code={quote(token, safe='')}"
125 if extra_params:
126 for key, value in extra_params.items():
127 params += f"&{quote(key, safe='')}={quote(value, safe='')}"
128
129 if "#" in return_url:
130 base_part, hash_part = return_url.split("#", 1)
131 return f"{_append_query_params(base_part, params)}#{hash_part}"
132
133 return _append_query_params(return_url, params)
134
135
136def _append_query_params(base: str, params: str) -> str:
137 """Append query params to a URL part, reusing a trailing ? or & separator if present."""
138 if base.endswith(("?", "&")):
139 return f"{base}{params}"
140 separator = "&" if "?" in base else "?"
141 return f"{base}{separator}{params}"
142
143
144def _is_private_ip(hostname: str) -> bool:
145 """Check if hostname is a private IP address (RFC 1918)."""
146 try:
147 ip = ipaddress.ip_address(hostname)
148 return ip.is_private
149 except ValueError:
150 # Not a valid IP address
151 return False
152