"""Which stream addresses the server may open for track titles (/radio/sse/, docs/radio.md). - Never an address inside a private network (loopback, 10.x, 172.16-31.x, 192.168.x, link-local, the Docker network …): the server would otherwise reach what only it can see. - Without an account only the directory's stations: the page's curated lists, the featured ones, and those radio-browser.info lists (by its station id, or by the address). Bringing an address of one's own is for saving it, and saving needs an account. """ import functools import hashlib import ipaddress import re import socket from pathlib import Path from urllib.parse import urlparse import requests from django.core.cache import cache from .models import FeaturedStation DIRECTORY = 'https://de1.api.radio-browser.info/json/stations' # The page's curated lists (SomaFM, SWR2 …) are part of the directory too; their addresses # are read from where the page has them, so they stand in one place only. CURATED_JS = Path(__file__).resolve().parent.parent / 'static' / 'js' / 'app' / '16-stations.js' DIRECTORY_TIMEOUT = 6 DIRECTORY_CACHE = 24 * 3600 def public_address(url): """True when url is http(s) and every address its host resolves to is a public one.""" try: parsed = urlparse(url) except ValueError: return False if parsed.scheme not in ('http', 'https') or not parsed.hostname: return False try: infos = socket.getaddrinfo(parsed.hostname, parsed.port or (443 if parsed.scheme == 'https' else 80), proto=socket.IPPROTO_TCP) except (OSError, UnicodeError, ValueError): return False addresses = {info[4][0] for info in infos} return bool(addresses) and all(ipaddress.ip_address(a.split('%')[0]).is_global for a in addresses) def _directory_lists(url, uuid): if uuid: try: r = requests.get(f'{DIRECTORY}/byuuid/{uuid}', timeout=DIRECTORY_TIMEOUT, headers={'User-Agent': 'diora/1.0'}) if r.ok and any(url in (s.get('url'), s.get('url_resolved')) for s in r.json()): return True except (requests.RequestException, ValueError): pass try: r = requests.get(f'{DIRECTORY}/byurl', params={'url': url}, timeout=DIRECTORY_TIMEOUT, headers={'User-Agent': 'diora/1.0'}) return bool(r.ok and r.json()) except (requests.RequestException, ValueError): return False @functools.lru_cache(maxsize=1) def curated(): try: return frozenset(re.findall(r"url: '([^']+)'", CURATED_JS.read_text(encoding='utf-8'))) except OSError: return frozenset() def in_directory(url, uuid='', own_host=''): """A station of the directory: curated on the page, featured, ours (the creamfresh stream through this server), or listed by radio-browser.info. Answers are kept a day.""" if url in curated() or FeaturedStation.objects.filter(url=url, active=True).exists(): return True if own_host and urlparse(url).netloc == own_host: return True key = 'radio:dir:' + hashlib.sha1(f'{url} {uuid}'.encode()).hexdigest() known = cache.get(key) if known is None: known = _directory_lists(url, uuid) cache.set(key, known, DIRECTORY_CACHE) return known