feat: lock playlist/shows endpoints behind Bearer token auth
The five internal endpoints (/playlist, /playlist/{position}, /shows,
/shows/by-episode/{ep}, /shows/{id}) now require admin authentication.
Dashboard JS, Sopel plugin, and Limnoria plugin updated to send the
token on GET requests. Six new 401 tests added.
Made-with: Cursor
This commit is contained in:
@@ -28,9 +28,12 @@ class ApiError(Exception):
|
||||
super().__init__(f"{status_code}: {detail}")
|
||||
|
||||
|
||||
def _api_get(base_url: str, path: str) -> dict:
|
||||
def _api_get(base_url: str, path: str, token: str = "") -> dict:
|
||||
url = f"{base_url.rstrip('/')}{path}"
|
||||
req = urllib.request.Request(url, headers={"Accept": "application/json"})
|
||||
headers = {"Accept": "application/json"}
|
||||
if token:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||
raw = resp.read().decode()
|
||||
@@ -234,8 +237,9 @@ class NtrPlaylist(callbacks.Plugin):
|
||||
if match:
|
||||
position = match.group(1)
|
||||
base_url = self.registryValue("apiBaseUrl")
|
||||
token = self.registryValue("adminToken")
|
||||
try:
|
||||
data = _api_get(base_url, f"/playlist/{position}")
|
||||
data = _api_get(base_url, f"/playlist/{position}", token)
|
||||
irc.reply(format_track(data))
|
||||
except ApiError as exc:
|
||||
LOGGER.warning("API error for !%s: %s", position, exc)
|
||||
@@ -260,8 +264,9 @@ class NtrPlaylist(callbacks.Plugin):
|
||||
irc.reply("Usage: !song <episode> <position>")
|
||||
return
|
||||
base_url = self.registryValue("apiBaseUrl")
|
||||
token = self.registryValue("adminToken")
|
||||
try:
|
||||
data = _api_get(base_url, f"/shows/by-episode/{episode}")
|
||||
data = _api_get(base_url, f"/shows/by-episode/{episode}", token)
|
||||
except ApiError as exc:
|
||||
LOGGER.warning("API error for !song %s %s: %s", episode, position, exc)
|
||||
irc.reply(exc.detail)
|
||||
@@ -280,6 +285,7 @@ class NtrPlaylist(callbacks.Plugin):
|
||||
Returns the playlist for the current show, or a specific episode.
|
||||
"""
|
||||
base_url = self.registryValue("apiBaseUrl")
|
||||
token = self.registryValue("adminToken")
|
||||
if text and text.strip():
|
||||
try:
|
||||
episode = int(text.strip())
|
||||
@@ -287,14 +293,14 @@ class NtrPlaylist(callbacks.Plugin):
|
||||
irc.reply("Usage: !playlist [episode]")
|
||||
return
|
||||
try:
|
||||
data = _api_get(base_url, f"/shows/by-episode/{episode}")
|
||||
data = _api_get(base_url, f"/shows/by-episode/{episode}", token)
|
||||
except ApiError as exc:
|
||||
LOGGER.warning("API error for playlist: %s", exc)
|
||||
irc.reply(exc.detail)
|
||||
return
|
||||
else:
|
||||
try:
|
||||
data = _api_get(base_url, "/playlist")
|
||||
data = _api_get(base_url, "/playlist", token)
|
||||
except ApiError as exc:
|
||||
LOGGER.warning("API error for playlist: %s", exc)
|
||||
irc.reply(exc.detail)
|
||||
@@ -316,8 +322,9 @@ class NtrPlaylist(callbacks.Plugin):
|
||||
irc.reply("Usage: !lastshow <position>")
|
||||
return
|
||||
base_url = self.registryValue("apiBaseUrl")
|
||||
token = self.registryValue("adminToken")
|
||||
try:
|
||||
shows = _api_get(base_url, "/shows?limit=2")
|
||||
shows = _api_get(base_url, "/shows?limit=2", token)
|
||||
except ApiError as exc:
|
||||
LOGGER.warning("API error for lastshow: %s", exc)
|
||||
irc.reply(exc.detail)
|
||||
@@ -327,7 +334,7 @@ class NtrPlaylist(callbacks.Plugin):
|
||||
return
|
||||
prev_show_id = shows[1]["id"]
|
||||
try:
|
||||
data = _api_get(base_url, f"/shows/{prev_show_id}")
|
||||
data = _api_get(base_url, f"/shows/{prev_show_id}", token)
|
||||
except ApiError as exc:
|
||||
LOGGER.warning("API error for lastshow show %s: %s", prev_show_id, exc)
|
||||
irc.reply(exc.detail)
|
||||
|
||||
@@ -143,9 +143,12 @@ class ApiError(Exception):
|
||||
super().__init__(f"{status_code}: {detail}")
|
||||
|
||||
|
||||
def _api_get(base_url: str, path: str) -> dict:
|
||||
def _api_get(base_url: str, path: str, token: str = "") -> dict:
|
||||
url = f"{base_url.rstrip('/')}{path}"
|
||||
req = urllib.request.Request(url, headers={"Accept": "application/json"})
|
||||
headers = {"Accept": "application/json"}
|
||||
if token:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||
raw = resp.read().decode()
|
||||
@@ -235,9 +238,10 @@ def _is_admin(bot, nick: str) -> bool:
|
||||
@plugin.rule(r"^!(\d+)$")
|
||||
def ntr_playlist_position(bot, trigger):
|
||||
base_url = bot.settings.ntr_playlist.api_base_url
|
||||
token = bot.settings.ntr_playlist.admin_token
|
||||
position = trigger.group(1)
|
||||
try:
|
||||
data = _api_get(base_url, f"/playlist/{position}")
|
||||
data = _api_get(base_url, f"/playlist/{position}", token)
|
||||
bot.say(format_track(data))
|
||||
except ApiError as e:
|
||||
LOGGER.warning("API error for !%s: %s", position, e)
|
||||
@@ -261,8 +265,9 @@ def ntr_song(bot, trigger):
|
||||
bot.say("Usage: !song <episode> <position>")
|
||||
return
|
||||
base_url = bot.settings.ntr_playlist.api_base_url
|
||||
token = bot.settings.ntr_playlist.admin_token
|
||||
try:
|
||||
data = _api_get(base_url, f"/shows/by-episode/{episode}")
|
||||
data = _api_get(base_url, f"/shows/by-episode/{episode}", token)
|
||||
except ApiError as e:
|
||||
LOGGER.warning("API error for !song %s %s: %s", episode, position, e)
|
||||
bot.say(e.detail)
|
||||
@@ -279,6 +284,7 @@ def ntr_song(bot, trigger):
|
||||
def ntr_playlist(bot, trigger):
|
||||
raw = trigger.group(2)
|
||||
base_url = bot.settings.ntr_playlist.api_base_url
|
||||
token = bot.settings.ntr_playlist.admin_token
|
||||
if raw and raw.strip():
|
||||
try:
|
||||
episode = int(raw.strip())
|
||||
@@ -286,14 +292,14 @@ def ntr_playlist(bot, trigger):
|
||||
bot.say("Usage: !playlist [episode]")
|
||||
return
|
||||
try:
|
||||
data = _api_get(base_url, f"/shows/by-episode/{episode}")
|
||||
data = _api_get(base_url, f"/shows/by-episode/{episode}", token)
|
||||
except ApiError as e:
|
||||
LOGGER.warning("API error for !playlist: %s", e)
|
||||
bot.say(e.detail)
|
||||
return
|
||||
else:
|
||||
try:
|
||||
data = _api_get(base_url, "/playlist")
|
||||
data = _api_get(base_url, "/playlist", token)
|
||||
except ApiError as e:
|
||||
LOGGER.warning("API error for !playlist: %s", e)
|
||||
bot.say(e.detail)
|
||||
@@ -313,8 +319,9 @@ def ntr_lastshow(bot, trigger):
|
||||
bot.say("Usage: !lastshow <position>")
|
||||
return
|
||||
base_url = bot.settings.ntr_playlist.api_base_url
|
||||
token = bot.settings.ntr_playlist.admin_token
|
||||
try:
|
||||
shows = _api_get(base_url, "/shows?limit=2")
|
||||
shows = _api_get(base_url, "/shows?limit=2", token)
|
||||
except ApiError as e:
|
||||
LOGGER.warning("API error for !lastshow: %s", e)
|
||||
bot.say(e.detail)
|
||||
@@ -324,7 +331,7 @@ def ntr_lastshow(bot, trigger):
|
||||
return
|
||||
prev_show_id = shows[1]["id"]
|
||||
try:
|
||||
data = _api_get(base_url, f"/shows/{prev_show_id}")
|
||||
data = _api_get(base_url, f"/shows/{prev_show_id}", token)
|
||||
except ApiError as e:
|
||||
LOGGER.warning("API error for !lastshow show %s: %s", prev_show_id, e)
|
||||
bot.say(e.detail)
|
||||
|
||||
Reference in New Issue
Block a user