Conform URLs (#281)

* Make any URLs build from env vars tolerant of path prefix, trailing/leading slashes

* Add comment

* Lint

Author: mderynck

engine/apps/integrations/metadata/heartbeat/_heartbeat_text_creator.py

@@ -1,8 +1,9 @@
 from dataclasses import dataclass
-from urllib.parse import urljoin
 
 from django.conf import settings
 
+from common.api_helpers.utils import create_engine_url
+
 
 @dataclass
 class IntegrationHeartBeatText:
@@ -31,7 +32,7 @@ def _get_heartbeat_expired_title(self):
         return heartbeat_expired_title
 
     def _get_heartbeat_expired_message(self):
-        heartbeat_docs_url = urljoin(settings.DOCS_URL, "/#/integrations/heartbeat")
+        heartbeat_docs_url = create_engine_url("/#/integrations/heartbeat", override_base=settings.DOCS_URL)
         heartbeat_expired_message = (
             f"Amixr was waiting for a heartbeat from {self.integration_verbal}. "
             f"Heartbeat is missing. That could happen because {self.integration_verbal} stopped or"

engine/apps/integrations/mixins/browsable_instruction_mixin.py

@@ -1,16 +1,17 @@
 import json
-from urllib.parse import urljoin
 
 from django.conf import settings
 from django.http import HttpResponse
 from django.template import loader
 
+from common.api_helpers.utils import create_engine_url
+
 
 class BrowsableInstructionMixin:
     def get(self, request, alert_receive_channel, *args, **kwargs):
         template = loader.get_template("integration_link.html")
         # TODO Create associative array for integrations
-        base_integration_docs_url = urljoin(settings.DOCS_URL, "/#/integrations/")
+        base_integration_docs_url = create_engine_url("/#/integrations/", override_base=settings.DOCS_URL)
         docs_url = f'{base_integration_docs_url}{request.get_full_path().split("/")[3]}'
         show_button = True
         if request.get_full_path().split("/")[3] == "amazon_sns":

engine/apps/integrations/views.py

@@ -1,6 +1,5 @@
 import json
 import logging
-from urllib.parse import urljoin
 
 from django.apps import apps
 from django.conf import settings
@@ -28,6 +27,7 @@
 from apps.integrations.tasks import create_alert, create_alertmanager_alerts
 from apps.sendgridapp.parse import Parse
 from apps.sendgridapp.permissions import AllowOnlySendgrid
+from common.api_helpers.utils import create_engine_url
 
 logger = logging.getLogger(__name__)
 
@@ -76,7 +76,7 @@ def handle_message(self, message, payload):
             raw_request_data = message
             title = message.get("AlarmName", "Alert")
         else:
-            docs_amazon_sns_url = urljoin(settings.DOCS_URL, "/#/integrations/amazon_sns")
+            docs_amazon_sns_url = create_engine_url("/#/integrations/amazon_sns", override_base=settings.DOCS_URL)
             title = "Alert"
             message_text = (
                 "Non-JSON payload received. Please make sure you publish monitoring Alarms to SNS,"
@@ -272,7 +272,7 @@ def post(self, request, alert_receive_channel, *args, **kwargs):
 class HeartBeatAPIView(AlertChannelDefiningMixin, APIView):
     def get(self, request, alert_receive_channel):
         template = loader.get_template("heartbeat_link.html")
-        docs_url = urljoin(settings.DOCS_URL, "/#/integrations/heartbeat")
+        docs_url = create_engine_url("/#/integrations/heartbeat", override_base=settings.DOCS_URL)
         return HttpResponse(
             template.render(
                 {

engine/apps/oss_installation/cloud_heartbeat.py

@@ -8,6 +8,7 @@
 from rest_framework import status
 
 from apps.base.utils import live_settings
+from common.api_helpers.utils import create_engine_url
 
 logger = logging.getLogger(__name__)
 
@@ -23,7 +24,7 @@ def setup_heartbeat_integration(name=None):
     # don't specify a team in the data, so heartbeat integration will be created in the General.
     name = name or f"OnCall Cloud Heartbeat {settings.BASE_URL}"
     data = {"type": "formatted_webhook", "name": name}
-    url = urljoin(settings.GRAFANA_CLOUD_ONCALL_API_URL, "api/v1/integrations/")
+    url = create_engine_url("api/v1/integrations/", override_base=settings.GRAFANA_CLOUD_ONCALL_API_URL)
     try:
         headers = {"Authorization": api_token}
         r = requests.post(url=url, data=data, headers=headers, timeout=5)

engine/apps/oss_installation/models/cloud_connector.py

@@ -7,6 +7,7 @@
 from apps.base.utils import live_settings
 from apps.oss_installation.models.cloud_user_identity import CloudUserIdentity
 from apps.user_management.models import User
+from common.api_helpers.utils import create_engine_url
 from common.constants.role import Role
 from settings.base import GRAFANA_CLOUD_ONCALL_API_URL
 
@@ -33,7 +34,7 @@ def sync_with_cloud(cls, token=None):
             logger.warning("Unable to sync with cloud. GRAFANA_CLOUD_ONCALL_TOKEN is not set")
             error_msg = "GRAFANA_CLOUD_ONCALL_TOKEN is not set"
         else:
-            info_url = urljoin(GRAFANA_CLOUD_ONCALL_API_URL, "api/v1/info/")
+            info_url = create_engine_url("api/v1/info/", override_base=GRAFANA_CLOUD_ONCALL_API_URL)
             try:
                 r = requests.get(info_url, headers={"AUTHORIZATION": api_token}, timeout=5)
                 if r.status_code == 200:
@@ -62,7 +63,7 @@ def sync_users_with_cloud(self) -> tuple[bool, str]:
 
         existing_emails = list(User.objects.filter(role__in=(Role.ADMIN, Role.EDITOR)).values_list("email", flat=True))
         matching_users = []
-        users_url = urljoin(GRAFANA_CLOUD_ONCALL_API_URL, "api/v1/users")
+        users_url = create_engine_url("api/v1/users", override_base=GRAFANA_CLOUD_ONCALL_API_URL)
 
         fetch_next_page = True
         users_fetched = True
@@ -115,7 +116,10 @@ def sync_user_with_cloud(self, user):
             logger.warning(f"Unable to sync_user_with cloud user_id {user.id}. GRAFANA_CLOUD_ONCALL_TOKEN is not set")
             error_msg = "GRAFANA_CLOUD_ONCALL_TOKEN is not set"
         else:
-            url = urljoin(GRAFANA_CLOUD_ONCALL_API_URL, f"api/v1/users/?email={user.email}&roles=0&roles=1&short=true")
+            url = create_engine_url(
+                f"api/v1/users/?email={user.email}&roles=0&roles=1&short=true",
+                override_base=GRAFANA_CLOUD_ONCALL_API_URL,
+            )
             try:
                 r = requests.get(url, headers={"AUTHORIZATION": api_token}, timeout=5)
                 if r.status_code != 200:

engine/apps/telegram/client.py

@@ -1,5 +1,4 @@
 from typing import Optional, Tuple, Union
-from urllib.parse import urljoin
 
 from telegram import Bot, InlineKeyboardMarkup, Message, ParseMode
 from telegram.error import InvalidToken, Unauthorized
@@ -10,6 +9,7 @@
 from apps.telegram.models import TelegramMessage
 from apps.telegram.renderers.keyboard import TelegramKeyboardRenderer
 from apps.telegram.renderers.message import TelegramMessageRenderer
+from common.api_helpers.utils import create_engine_url
 
 
 class TelegramClient:
@@ -34,7 +34,7 @@ def is_chat_member(self, chat_id: Union[int, str]) -> bool:
             return False
 
     def register_webhook(self, webhook_url: Optional[str] = None) -> None:
-        webhook_url = webhook_url or urljoin(live_settings.TELEGRAM_WEBHOOK_HOST, "/telegram/")
+        webhook_url = webhook_url or create_engine_url("/telegram/", override_base=live_settings.TELEGRAM_WEBHOOK_HOST)
 
         if webhook_url is None:
             webhook_url = live_settings.TELEGRAM_WEBHOOK_URL

engine/apps/twilioapp/models/phone_call.py

@@ -1,5 +1,4 @@
 import logging
-from urllib.parse import urljoin
 
 import requests
 from django.apps import apps
@@ -14,6 +13,7 @@
 from apps.base.utils import live_settings
 from apps.twilioapp.constants import TwilioCallStatuses
 from apps.twilioapp.twilio_client import twilio_client
+from common.api_helpers.utils import create_engine_url
 from common.utils import clean_markup, escape_for_twilio_phone_call
 
 logger = logging.getLogger(__name__)
@@ -158,7 +158,7 @@ def created_for_slack(self):
 
     @classmethod
     def _make_cloud_call(cls, user, message_body):
-        url = urljoin(settings.GRAFANA_CLOUD_ONCALL_API_URL, "api/v1/make_call")
+        url = create_engine_url("api/v1/make_call", override_base=settings.GRAFANA_CLOUD_ONCALL_API_URL)
         auth = {"Authorization": live_settings.GRAFANA_CLOUD_ONCALL_TOKEN}
         data = {
             "email": user.email,

engine/apps/twilioapp/models/sms_message.py

@@ -1,5 +1,4 @@
 import logging
-from urllib.parse import urljoin
 
 import requests
 from django.apps import apps
@@ -13,6 +12,7 @@
 from apps.base.utils import live_settings
 from apps.twilioapp.constants import TwilioMessageStatuses
 from apps.twilioapp.twilio_client import twilio_client
+from common.api_helpers.utils import create_engine_url
 from common.utils import clean_markup
 
 logger = logging.getLogger(__name__)
@@ -123,7 +123,7 @@ def created_for_slack(self):
 
     @classmethod
     def _send_cloud_sms(cls, user, message_body):
-        url = urljoin(settings.GRAFANA_CLOUD_ONCALL_API_URL, "api/v1/send_sms")
+        url = create_engine_url("api/v1/send_sms", override_base=settings.GRAFANA_CLOUD_ONCALL_API_URL)
         auth = {"Authorization": live_settings.GRAFANA_CLOUD_ONCALL_TOKEN}
         data = {
             "email": user.email,

engine/common/api_helpers/utils.py

@@ -54,6 +54,15 @@ def validate_ical_url(url):
     return None
 
 
+"""
+This utility function is for building a URL when we don't know if the base URL
+has been given a trailing / such as reading from environment variable or user
+input.  If the base URL is coming from a validated model field urljoin can be used
+instead.  Do not use this function to append query parameters since a / is added
+to the end of the base URL if there isn't one.
+"""
+
+
 def create_engine_url(path, override_base=None):
     base = settings.BASE_URL
     if override_base: