84 lines
3.3 KiB
Python
84 lines
3.3 KiB
Python
"""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
|