centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Conform URLs (#281)

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

* Add comment

* Lint
This commit is contained in:
Michael Derynck 2022-07-25 09:12:50 -06:00 • committed by GitHub
parent 76f67b171a
commit ce8f4e53fa
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 33 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
engine/apps/integrations/metadata/heartbeat/_heartbeat_text_creator.py
View file

@ -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 @@ class HeartBeatTextCreator:
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"

5
engine/apps/integrations/mixins/browsable_instruction_mixin.py
View file

@ -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":

6
engine/apps/integrations/views.py
View file

@ -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.mixins import (
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 @@ class AmazonSNS(BrowsableInstructionMixin, SNSEndpoint):
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 @@ class UniversalAPIView(BrowsableInstructionMixin, AlertChannelDefiningMixin, Int
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(
{

3
engine/apps/oss_installation/cloud_heartbeat.py
View file

@ -8,6 +8,7 @@ from django.conf import settings
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)

10
engine/apps/oss_installation/models/cloud_connector.py
View file

@ -7,6 +7,7 @@ from django.db import models, transaction
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 @@ class CloudConnector(models.Model):
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 @@ class CloudConnector(models.Model):
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 @@ class CloudConnector(models.Model):
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:

4
engine/apps/telegram/client.py
View file

@ -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.base.utils import live_settings
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 @@ class TelegramClient:
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

4
engine/apps/twilioapp/models/phone_call.py
View file

@ -1,5 +1,4 @@
import logging
from urllib.parse import urljoin
import requests
from django.apps import apps
@ -14,6 +13,7 @@ from apps.alerts.signals import user_notification_action_triggered_signal
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 @@ class PhoneCall(models.Model):
@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,

4
engine/apps/twilioapp/models/sms_message.py
View file

@ -1,5 +1,4 @@
import logging
from urllib.parse import urljoin
import requests
from django.apps import apps
@ -13,6 +12,7 @@ from apps.alerts.signals import user_notification_action_triggered_signal
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 @@ class SMSMessage(models.Model):
@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,

9
engine/common/api_helpers/utils.py
View file

@ -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: