6700616467

Committed 31 Oct 2023 01:00AM UTC coverage: 92.452% (-1.2%) from 93.66%

Build # 6700616467

Build Type

Pull #383

github

Committed by

web-flow

Commit Message

Merge 300d02535 into fb838f71e

Pull Request Pull Request #383: Add offboarding sequences

Run Details

6161 of 6664 relevant lines covered (92.45%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.36

back/admin/integrations/models.py

import time
import base64
import io
import json
import uuid
from datetime import timedelta

import requests

from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import models
from django.db.models.signals import post_delete
from django.dispatch import receiver
from django.template import Context, Template
from django.urls import reverse_lazy
from django.utils import timezone
from django.utils.crypto import get_random_string
from django.utils.translation import gettext_lazy as _
from django_q.models import Schedule
from django_q.tasks import schedule
from requests.exceptions import (
    HTTPError,
    InvalidHeader,
    InvalidJSONError,
    InvalidSchema,
    InvalidURL,
    JSONDecodeError,
    MissingSchema,
    SSLError,
    Timeout,
    TooManyRedirects,
    URLRequired,
)
from twilio.rest import Client

from admin.integrations.utils import get_value_from_notation
from admin.integrations.serializers import (
    WebhookManifestSerializer,
    SyncUsersManifestSerializer,
)
from misc.fernet_fields import EncryptedTextField
from misc.fields import EncryptedJSONField
from organization.models import Notification
from organization.utils import send_email_with_notification


class IntegrationManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset()

    def sequence_integration_options(self):
        # any webhooks and account provisioning
        return self.get_queryset().filter(
            integration=Integration.Type.CUSTOM,
            manifest_type__in=[
                Integration.ManifestType.WEBHOOK,
                Integration.ManifestType.MANUAL_USER_PROVISIONING,
            ],
        )

    def account_provision_options(self):
        # only account provisioning (no general webhooks)
        return self.get_queryset().filter(
            integration=Integration.Type.CUSTOM,
            manifest_type=Integration.ManifestType.WEBHOOK,
            manifest__exists__isnull=False,
        ) | self.get_queryset().filter(
            integration=Integration.Type.CUSTOM,
            manifest_type=Integration.ManifestType.MANUAL_USER_PROVISIONING,
        )

    def import_users_options(self):
        # only import user items
        return (
            self.get_queryset()
            .filter(
                integration=Integration.Type.CUSTOM,
                manifest_type=Integration.ManifestType.SYNC_USERS,
            )
            .exclude(manifest__schedule__isnull=False)
        )


class Integration(models.Model):
    class Type(models.IntegerChoices):
        SLACK_BOT = 0, _("Slack bot")
        SLACK_ACCOUNT_CREATION = 1, _("Slack account creation")  # legacy
        GOOGLE_ACCOUNT_CREATION = 2, _("Google account creation")  # legacy
        GOOGLE_LOGIN = 3, _("Google Login")
        ASANA = 4, _("Asana")  # legacy
        CUSTOM = 10, _("Custom")

    class ManifestType(models.IntegerChoices):
        WEBHOOK = 0, _("Provision user accounts or trigger webhooks")
        SYNC_USERS = 1, _("Sync users")
        MANUAL_USER_PROVISIONING = 3, _(
            "Manual user account provisioning, no manifest required"
        )

    name = models.CharField(max_length=300, default="", blank=True)
    integration = models.IntegerField(choices=Type.choices)
    manifest_type = models.IntegerField(
        choices=ManifestType.choices, null=True, blank=True
    )
    token = EncryptedTextField(max_length=10000, default="", blank=True)
    refresh_token = EncryptedTextField(max_length=10000, default="", blank=True)
    base_url = models.CharField(max_length=22300, default="", blank=True)
    redirect_url = models.CharField(max_length=22300, default="", blank=True)
    account_id = models.CharField(max_length=22300, default="", blank=True)
    active = models.BooleanField(default=True)
    ttl = models.IntegerField(null=True, blank=True)
    expiring = models.DateTimeField(auto_now_add=True, blank=True)
    one_time_auth_code = models.UUIDField(
        default=uuid.uuid4, editable=False, unique=True
    )

    manifest = models.JSONField(default=dict, null=True, blank=True)
    extra_args = EncryptedJSONField(default=dict)
    enabled_oauth = models.BooleanField(default=False)

    # Slack
    app_id = models.CharField(max_length=100, default="")
    client_id = models.CharField(max_length=100, default="")
    client_secret = models.CharField(max_length=100, default="")
    signing_secret = models.CharField(max_length=100, default="")
    verification_token = models.CharField(max_length=100, default="")
    bot_token = EncryptedTextField(max_length=10000, default="", blank=True)
    bot_id = models.CharField(max_length=100, default="")

    @property
    def skip_user_provisioning(self):
        return self.manifest_type == Integration.ManifestType.MANUAL_USER_PROVISIONING

    @property
    def schedule_name(self):
        return f"User sync for integration: {self.id}"

    def clean(self):
        if not self.manifest or self.skip_user_provisioning:
            # ignore field if form doesn't have it or no manifest is necessary
            return

        if self.manifest_type == Integration.ManifestType.WEBHOOK:
            manifest_serializer = WebhookManifestSerializer(data=self.manifest)
        else:
            manifest_serializer = SyncUsersManifestSerializer(data=self.manifest)
        if not manifest_serializer.is_valid():
            raise ValidationError({"manifest": json.dumps(manifest_serializer.errors)})

    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)

        # skip if it's not a sync user integration (no background jobs for the others)
        if self.manifest_type != Integration.ManifestType.SYNC_USERS:
            return

        # update the background job based on the manifest
        schedule_cron = self.manifest.get("schedule")

        try:
            schedule_obj = Schedule.objects.get(name=self.schedule_name)
        except Schedule.DoesNotExist:
            # Schedule does not exist yet, so create it if specified
            if schedule_cron:
                schedule(
                    "admin.integrations.tasks.sync_user_info",
                    self.id,
                    schedule_type=Schedule.CRON,
                    cron=schedule_cron,
                    name=self.schedule_name,
                )
            return

        # delete if cron was removed
        if schedule_cron is None:
            schedule_obj.delete()
            return

        # if schedule changed, then update
        if schedule_obj.cron != schedule_cron:
            schedule_obj.cron = schedule_cron
            schedule_obj.save()

    def run_request(self, data):
        url = self._replace_vars(data["url"])
        if "data" in data:
            post_data = self._replace_vars(json.dumps(data["data"]))
        else:
            post_data = {}
        if data.get("cast_data_to_json", False):
            try:
                post_data = json.loads(post_data)
            except Exception:
                pass

        # extract files from locally saved files and send them with the request
        files_to_send = {}
        for field_name, file_name in data.get("files", {}).items():
            try:
                files_to_send[field_name] = (file_name, self.params["files"][file_name])
            except KeyError:
                return (
                    False,
                    f"{file_name} could not be found in the locally saved files",
                )

        try:
            response = requests.request(
                data.get("method", "POST"),
                url,
                headers=self.headers(data.get("headers", {})),
                data=post_data,
                files=files_to_send,
                timeout=120,
            )
        except (InvalidJSONError, JSONDecodeError):
            return False, "JSON is invalid"

        except HTTPError:
            return False, "An HTTP error occurred"

        except SSLError:
            return False, "An SSL error occurred"

        except Timeout:
            return False, "The request timed out"

        except (URLRequired, MissingSchema, InvalidSchema, InvalidURL):
            return False, "The url is invalid"

        except TooManyRedirects:
            return False, "There are too many redirects"

        except InvalidHeader:
            return False, "The header is invalid"

        except:  # noqa E722
            return False, "There was an unexpected error with the request"

        if data.get("fail_when_4xx_response_code", True):
            try:
                response.raise_for_status()
            except Exception:
                return False, response.text

        return True, response

    def _replace_vars(self, text):
        params = {} if not hasattr(self, "params") else self.params
        params["redirect_url"] = settings.BASE_URL + reverse_lazy(
            "integrations:oauth-callback", args=[self.id]
        )
        if hasattr(self, "new_hire") and self.new_hire is not None:
            text = self.new_hire.personalize(text, self.extra_args | params)
            return text
        t = Template(text)
        context = Context(self.extra_args | params)
        text = t.render(context)
        return text

    @property
    def has_oauth(self):
        return "oauth" in self.manifest

    def headers(self, headers=None):
        if headers is None:
            headers = {}

        headers = (
            self.manifest.get("headers", {}).items()
            if len(headers) == 0
            else headers.items()
        )
        new_headers = {}
        for key, value in headers:
            # If Basic authentication then swap to base64
            if key == "Authorization" and value.startswith("Basic"):
                auth_details = self._replace_vars(value.split(" ", 1)[1])
                value = "Basic " + base64.b64encode(
                    auth_details.encode("ascii")
                ).decode("ascii")

            # Adding an empty string to force to return a string instead of a
            # safestring. Ref: https://github.com/psf/requests/issues/6159
            new_headers[self._replace_vars(key) + ""] = self._replace_vars(value) + ""
        return new_headers

    def user_exists(self, new_hire):
        # check if user has been created manually
        if self.skip_user_provisioning:
            from users.models import IntegrationUser

            try:
                user_integration = IntegrationUser.objects.get(
                    user=new_hire, integration=self
                )
            except IntegrationUser.DoesNotExist:
                return False

            return not user_integration.revoked

        self.new_hire = new_hire
        self.has_user_context = new_hire is not None

        # Renew token if necessary
        if not self.renew_key():
            return

        success, response = self.run_request(self.manifest["exists"])

        if not success:
            return None

        return self._replace_vars(self.manifest["exists"]["expected"]) in response.text

    def needs_user_info(self, user):
        if self.skip_user_provisioning:
            return False

        # form created from the manifest, this info is always needed to create a new
        # account. Check if there is anything that needs to be filled
        form = self.manifest.get("form", [])

        # extra items that are needed from the integration (often prefilled by admin)
        extra_user_info = self.manifest.get("extra_user_info", [])
        needs_more_info = any(
            item["id"] not in user.extra_fields.keys() for item in extra_user_info
        )

        return len(form) > 0 or needs_more_info

    def revoke_user(self, user):
        if self.skip_user_provisioning:
            # should never be triggered
            return False, "Cannot revoke manual integration"

        self.new_hire = user
        self.has_user_context = True

        # Renew token if necessary
        if not self.renew_key():
            return False, "Couldn't renew key"

        revoke_manifest = self.manifest.get("revoke", [])

        # add extra fields directly to params
        self.params = self.new_hire.extra_fields

        for item in revoke_manifest:
            success, response = self.run_request(item)

            if not success:
                return False, self.clean_response(response)

        return True, ""

    def renew_key(self):
        # Oauth2 refreshing access token if needed
        success = True
        if (
            self.has_oauth
            and "expires_in" in self.extra_args.get("oauth", {})
            and self.expiring < timezone.now()
        ):
            success, response = self.run_request(self.manifest["oauth"]["refresh"])

            if not success:
                user = self.new_hire if self.has_user_context else None
                Notification.objects.create(
                    notification_type=Notification.Type.FAILED_INTEGRATION,
                    extra_text=self.name,
                    created_for=user,
                    description="Refresh url: " + str(response),
                )
                return success

            self.extra_args["oauth"] |= response.json()
            if "expires_in" in response.json():
                self.expiring = timezone.now() + timedelta(
                    seconds=response.json()["expires_in"]
                )
            self.save(update_fields=["expiring", "extra_args"])
        return success

    def _check_condition(self, response, condition):
        value = self._replace_vars(condition.get("value"))
        try:
            # first argument will be taken from the response
            response_value = get_value_from_notation(
                condition.get("response_notation"), response.json()
            )
        except KeyError:
            # we know that the result might not be in the response yet, as we are
            # waiting for the correct response, so just respond with an empty string
            response_value = ""
        return value == response_value

    def _polling(self, item, response):
        polling = item.get("polling")
        continue_if = item.get("continue_if")
        interval = polling.get("interval")
        amount = polling.get("amount")

        got_expected_result = self._check_condition(response, continue_if)
        if got_expected_result:
            return True, response

        tried = 1
        while amount > tried:
            time.sleep(interval)
            success, response = self.run_request(item)
            got_expected_result = self._check_condition(response, continue_if)
            if got_expected_result:
                return True, response
            tried += 1
        # if exceeding the max amounts, then fail
        return False, response

    def execute(self, new_hire=None, params=None):
        self.params = params or {}
        self.params["responses"] = []
        self.params["files"] = {}
        self.new_hire = new_hire
        self.has_user_context = new_hire is not None

        if self.has_user_context:
            self.params |= new_hire.extra_fields
            self.new_hire = new_hire

        # Renew token if necessary
        if not self.renew_key():
            return False, None

        # Add generated secrets
        for item in self.manifest.get("initial_data_form", []):
            if "name" in item and item["name"] == "generate":
                self.extra_args[item["id"]] = get_random_string(length=10)

        # Run all requests
        for item in self.manifest["execute"]:
            success, response = self.run_request(item)

            # check if we need to poll before continuing
            if polling := item.get("polling", False):
                success, response = self._polling(item, response)

            # check if we need to block this integration based on condition
            if continue_if := item.get("continue_if", False):
                got_expected_result = self._check_condition(response, continue_if)
                if not got_expected_result:
                    response = self.clean_response(response=response)
                    Notification.objects.create(
                        notification_type=Notification.Type.BLOCKED_INTEGRATION,
                        extra_text=self.name,
                        created_for=new_hire,
                        description=f"Execute url ({item['url']}): {response}",
                    )
                    return False, response

            # No need to retry or log when we are importing users
            if not success:
                if self.has_user_context:
                    response = self.clean_response(response=response)
                    if polling:
                        response = "Polling timed out: " + response
                    Notification.objects.create(
                        notification_type=Notification.Type.FAILED_INTEGRATION,
                        extra_text=self.name,
                        created_for=new_hire,
                        description=f"Execute url ({item['url']}): {response}",
                    )
                # Retry url in one hour
                try:
                    schedule(
                        "admin.integrations.tasks.retry_integration",
                        new_hire.id,
                        self.id,
                        params,
                        name=(
                            f"Retrying integration {self.id} for new hire {new_hire.id}"
                        ),
                        next_run=timezone.now() + timedelta(hours=1),
                        schedule_type=Schedule.ONCE,
                    )
                except:  # noqa E722
                    # Only errors when item gets added another time, so we can safely
                    # let it pass.
                    pass
                return False, response

            # save if file, so we can reuse later
            save_as_file = item.get("save_as_file")
            if save_as_file is not None:
                self.params["files"][save_as_file] = io.BytesIO(response.content)

            # save json response temporarily to be reused in other parts
            try:
                self.params["responses"].append(response.json())
            except:  # noqa E722
                # if we save a file, then just append an empty dict
                self.params["responses"].append({})

            # store data coming back from response to the user, so we can reuse in other
            # integrations
            if store_data := item.get("store_data", {}):
                for new_hire_prop, notation_for_response in store_data.items():
                    try:
                        value = get_value_from_notation(
                            notation_for_response, response.json()
                        )
                    except KeyError:
                        return (
                            False,
                            f"Could not store data to new hire: {notation_for_response}"
                            f" not found in {self.clean_response(response.json())}",
                        )

                    # save to new hire and to temp var `params` on this model for use in
                    # the same integration
                    new_hire.extra_fields[new_hire_prop] = value
                    self.params[new_hire_prop] = value
                new_hire.save()

        # Run all post requests (notifications)
        for item in self.manifest.get("post_execute_notification", []):
            if item["type"] == "email":
                send_email_with_notification(
                    subject=self._replace_vars(item["subject"]),
                    message=self._replace_vars(item["message"]),
                    to=self._replace_vars(item["to"]),
                    notification_type=(
                        Notification.Type.SENT_EMAIL_INTEGRATION_NOTIFICATION
                    ),
                )
                return True, None
            else:
                try:
                    client = Client(
                        settings.TWILIO_ACCOUNT_SID, settings.TWILIO_AUTH_TOKEN
                    )
                    client.messages.create(
                        to=new_hire.phone,
                        from_=settings.TWILIO_FROM_NUMBER,
                        body=self._replace_vars(item["message"]),
                    )
                except Exception:
                    Notification.objects.create(
                        notification_type=(
                            Notification.Type.FAILED_TEXT_INTEGRATION_NOTIFICATION
                        ),
                        extra_text=self.name,
                        created_for=new_hire,
                    )
                    return True, None

        # Succesfully ran integration, add notification only when we are provisioning
        # access
        if self.has_user_context:
            Notification.objects.create(
                notification_type=Notification.Type.RAN_INTEGRATION,
                extra_text=self.name,
                created_for=new_hire,
            )
        return True, response

    def config_form(self, data=None):
        if self.skip_user_provisioning:
            from .forms import ManualIntegrationConfigForm

            return ManualIntegrationConfigForm(data=data)

        else:
            from .forms import IntegrationConfigForm

            return IntegrationConfigForm(instance=self, data=data)

    def clean_response(self, response):
        # if json, then convert to string to make it easier to replace values
        response = str(response)
        for name, value in self.extra_args.items():
            response = response.replace(
                str(value), _("***Secret value for %(name)s***") % {"name": name}
            )

        return response

    objects = IntegrationManager()


@receiver(post_delete, sender=Integration)
def delete_schedule(sender, instance, **kwargs):
    Schedule.objects.filter(name=instance.schedule_name).delete()

1	import time	1✔
2	import base64	1✔
3	import io	1✔
4	import json	1✔
5	import uuid	1✔
6	from datetime import timedelta	1✔
7
8	import requests	1✔
9
10	from django.conf import settings	1✔
11	from django.core.exceptions import ValidationError	1✔
12	from django.db import models	1✔
13	from django.db.models.signals import post_delete	1✔
14	from django.dispatch import receiver	1✔
15	from django.template import Context, Template	1✔
16	from django.urls import reverse_lazy	1✔
17	from django.utils import timezone	1✔
18	from django.utils.crypto import get_random_string	1✔
19	from django.utils.translation import gettext_lazy as _	1✔
20	from django_q.models import Schedule	1✔
21	from django_q.tasks import schedule	1✔
22	from requests.exceptions import (	1✔
23	HTTPError,
24	InvalidHeader,
25	InvalidJSONError,
26	InvalidSchema,
27	InvalidURL,
28	JSONDecodeError,
29	MissingSchema,
30	SSLError,
31	Timeout,
32	TooManyRedirects,
33	URLRequired,
34	)
35	from twilio.rest import Client	1✔
36
37	from admin.integrations.utils import get_value_from_notation	1✔
38	from admin.integrations.serializers import (	1✔
39	WebhookManifestSerializer,
40	SyncUsersManifestSerializer,
41	)
42	from misc.fernet_fields import EncryptedTextField	1✔
43	from misc.fields import EncryptedJSONField	1✔
44	from organization.models import Notification	1✔
45	from organization.utils import send_email_with_notification	1✔
46
47
48	class IntegrationManager(models.Manager):	1✔
49	def get_queryset(self):	1✔
50	return super().get_queryset()	1✔
51
52	def sequence_integration_options(self):	1✔
53	# any webhooks and account provisioning
54	return self.get_queryset().filter(	1✔
55	integration=Integration.Type.CUSTOM,
56	manifest_type__in=[
57	Integration.ManifestType.WEBHOOK,
58	Integration.ManifestType.MANUAL_USER_PROVISIONING,
59	],
60	)
61
62	def account_provision_options(self):	1✔
63	# only account provisioning (no general webhooks)
64	return self.get_queryset().filter(	1✔
65	integration=Integration.Type.CUSTOM,
66	manifest_type=Integration.ManifestType.WEBHOOK,
67	manifest__exists__isnull=False,
68	) \| self.get_queryset().filter(
69	integration=Integration.Type.CUSTOM,
70	manifest_type=Integration.ManifestType.MANUAL_USER_PROVISIONING,
71	)
72
73	def import_users_options(self):	1✔
74	# only import user items
75	return (	1✔
76	self.get_queryset()
77	.filter(
78	integration=Integration.Type.CUSTOM,
79	manifest_type=Integration.ManifestType.SYNC_USERS,
80	)
81	.exclude(manifest__schedule__isnull=False)
82	)
83
84
85	class Integration(models.Model):	1✔
86	class Type(models.IntegerChoices):	1✔
87	SLACK_BOT = 0, _("Slack bot")	1✔
88	SLACK_ACCOUNT_CREATION = 1, _("Slack account creation") # legacy	1✔
89	GOOGLE_ACCOUNT_CREATION = 2, _("Google account creation") # legacy	1✔
90	GOOGLE_LOGIN = 3, _("Google Login")	1✔
91	ASANA = 4, _("Asana") # legacy	1✔
92	CUSTOM = 10, _("Custom")	1✔
93
94	class ManifestType(models.IntegerChoices):	1✔
95	WEBHOOK = 0, _("Provision user accounts or trigger webhooks")	1✔
96	SYNC_USERS = 1, _("Sync users")	1✔
97	MANUAL_USER_PROVISIONING = 3, _(	1✔
98	"Manual user account provisioning, no manifest required"
99	)
100
101	name = models.CharField(max_length=300, default="", blank=True)	1✔
102	integration = models.IntegerField(choices=Type.choices)	1✔
103	manifest_type = models.IntegerField(	1✔
104	choices=ManifestType.choices, null=True, blank=True
105	)
106	token = EncryptedTextField(max_length=10000, default="", blank=True)	1✔
107	refresh_token = EncryptedTextField(max_length=10000, default="", blank=True)	1✔
108	base_url = models.CharField(max_length=22300, default="", blank=True)	1✔
109	redirect_url = models.CharField(max_length=22300, default="", blank=True)	1✔
110	account_id = models.CharField(max_length=22300, default="", blank=True)	1✔
111	active = models.BooleanField(default=True)	1✔
112	ttl = models.IntegerField(null=True, blank=True)	1✔
113	expiring = models.DateTimeField(auto_now_add=True, blank=True)	1✔
114	one_time_auth_code = models.UUIDField(	1✔
115	default=uuid.uuid4, editable=False, unique=True
116	)
117
118	manifest = models.JSONField(default=dict, null=True, blank=True)	1✔
119	extra_args = EncryptedJSONField(default=dict)	1✔
120	enabled_oauth = models.BooleanField(default=False)	1✔
121
122	# Slack
123	app_id = models.CharField(max_length=100, default="")	1✔
124	client_id = models.CharField(max_length=100, default="")	1✔
125	client_secret = models.CharField(max_length=100, default="")	1✔
126	signing_secret = models.CharField(max_length=100, default="")	1✔
127	verification_token = models.CharField(max_length=100, default="")	1✔
128	bot_token = EncryptedTextField(max_length=10000, default="", blank=True)	1✔
129	bot_id = models.CharField(max_length=100, default="")	1✔
130
131	@property	1✔
132	def skip_user_provisioning(self):	1✔
133	return self.manifest_type == Integration.ManifestType.MANUAL_USER_PROVISIONING	1✔
134
135	@property	1✔
136	def schedule_name(self):	1✔
137	return f"User sync for integration: {self.id}"	1✔
138
139	def clean(self):	1✔
140	if not self.manifest or self.skip_user_provisioning:	1✔
141	# ignore field if form doesn't have it or no manifest is necessary
142	return	1✔
143
144	if self.manifest_type == Integration.ManifestType.WEBHOOK:	1✔
145	manifest_serializer = WebhookManifestSerializer(data=self.manifest)	1✔
146	else:
147	manifest_serializer = SyncUsersManifestSerializer(data=self.manifest)	1✔
148	if not manifest_serializer.is_valid():	1✔
149	raise ValidationError({"manifest": json.dumps(manifest_serializer.errors)})	×
150
151	def save(self, args, *kwargs):	1✔
152	super().save(args, *kwargs)	1✔
153
154	# skip if it's not a sync user integration (no background jobs for the others)
155	if self.manifest_type != Integration.ManifestType.SYNC_USERS:	1✔
156	return	1✔
157
158	# update the background job based on the manifest
159	schedule_cron = self.manifest.get("schedule")	1✔
160
161	try:	1✔
162	schedule_obj = Schedule.objects.get(name=self.schedule_name)	1✔
163	except Schedule.DoesNotExist:	1✔
164	# Schedule does not exist yet, so create it if specified
165	if schedule_cron:	1✔
166	schedule(	1✔
167	"admin.integrations.tasks.sync_user_info",
168	self.id,
169	schedule_type=Schedule.CRON,
170	cron=schedule_cron,
171	name=self.schedule_name,
172	)
173	return	1✔
174
175	# delete if cron was removed
176	if schedule_cron is None:	1✔
177	schedule_obj.delete()	1✔
178	return	1✔
179
180	# if schedule changed, then update
181	if schedule_obj.cron != schedule_cron:	×
182	schedule_obj.cron = schedule_cron	×
183	schedule_obj.save()	×
184
185	def run_request(self, data):	1✔
186	url = self._replace_vars(data["url"])	1✔
187	if "data" in data:	1✔
188	post_data = self._replace_vars(json.dumps(data["data"]))	1✔
189	else:
190	post_data = {}	1✔
191	if data.get("cast_data_to_json", False):	1✔
192	try:	×
193	post_data = json.loads(post_data)	×
194	except Exception:	×
195	pass	×
196
197	# extract files from locally saved files and send them with the request
198	files_to_send = {}	1✔
199	for field_name, file_name in data.get("files", {}).items():	1✔
200	try:	1✔
201	files_to_send[field_name] = (file_name, self.params["files"][file_name])	1✔
202	except KeyError:	1✔
203	return (	1✔
204	False,
205	f"{file_name} could not be found in the locally saved files",
206	)
207
208	try:	1✔
209	response = requests.request(	1✔
210	data.get("method", "POST"),
211	url,
212	headers=self.headers(data.get("headers", {})),
213	data=post_data,
214	files=files_to_send,
215	timeout=120,
216	)
217	except (InvalidJSONError, JSONDecodeError):	1✔
218	return False, "JSON is invalid"	×
219
220	except HTTPError:	1✔
221	return False, "An HTTP error occurred"	×
222
223	except SSLError:	1✔
224	return False, "An SSL error occurred"	×
225
226	except Timeout:	1✔
227	return False, "The request timed out"	×
228
229	except (URLRequired, MissingSchema, InvalidSchema, InvalidURL):	1✔
230	return False, "The url is invalid"	1✔
231
232	except TooManyRedirects:	×
233	return False, "There are too many redirects"	×
234
235	except InvalidHeader:	×
236	return False, "The header is invalid"	×
237
238	except: # noqa E722	×
239	return False, "There was an unexpected error with the request"	×
240
241	if data.get("fail_when_4xx_response_code", True):	1✔
242	try:	1✔
243	response.raise_for_status()	1✔
244	except Exception:	1✔
245	return False, response.text	1✔
246
247	return True, response	1✔
248
249	def _replace_vars(self, text):	1✔
250	params = {} if not hasattr(self, "params") else self.params	1✔
251	params["redirect_url"] = settings.BASE_URL + reverse_lazy(	1✔
252	"integrations:oauth-callback", args=[self.id]
253	)
254	if hasattr(self, "new_hire") and self.new_hire is not None:	1✔
255	text = self.new_hire.personalize(text, self.extra_args \| params)	1✔
256	return text	1✔
257	t = Template(text)	1✔
258	context = Context(self.extra_args \| params)	1✔
259	text = t.render(context)	1✔
260	return text	1✔
261
262	@property	1✔
263	def has_oauth(self):	1✔
264	return "oauth" in self.manifest	1✔
265
266	def headers(self, headers=None):	1✔
267	if headers is None:	1✔
268	headers = {}	1✔
269
270	headers = (	1✔
271	self.manifest.get("headers", {}).items()
272	if len(headers) == 0
273	else headers.items()
274	)
275	new_headers = {}	1✔
276	for key, value in headers:	1✔
277	# If Basic authentication then swap to base64
278	if key == "Authorization" and value.startswith("Basic"):	1✔
279	auth_details = self._replace_vars(value.split(" ", 1)[1])	1✔
280	value = "Basic " + base64.b64encode(	1✔
281	auth_details.encode("ascii")
282	).decode("ascii")
283
284	# Adding an empty string to force to return a string instead of a
285	# safestring. Ref: https://github.com/psf/requests/issues/6159
286	new_headers[self._replace_vars(key) + ""] = self._replace_vars(value) + ""	1✔
287	return new_headers	1✔
288
289	def user_exists(self, new_hire):	1✔
290	# check if user has been created manually
291	if self.skip_user_provisioning:	1✔
292	from users.models import IntegrationUser	1✔
293
294	try:	1✔
295	user_integration = IntegrationUser.objects.get(	1✔
296	user=new_hire, integration=self
297	)
298	except IntegrationUser.DoesNotExist:	1✔
299	return False	1✔
300
301	return not user_integration.revoked	1✔
302
303	self.new_hire = new_hire	1✔
304	self.has_user_context = new_hire is not None	1✔
305
306	# Renew token if necessary
307	if not self.renew_key():	1✔
308	return	×
309
310	success, response = self.run_request(self.manifest["exists"])	1✔
311
312	if not success:	1✔
313	return None	1✔
314
315	return self._replace_vars(self.manifest["exists"]["expected"]) in response.text	1✔
316
317	def needs_user_info(self, user):	1✔
318	if self.skip_user_provisioning:	1✔
319	return False	1✔
320
321	# form created from the manifest, this info is always needed to create a new
322	# account. Check if there is anything that needs to be filled
323	form = self.manifest.get("form", [])	1✔
324
325	# extra items that are needed from the integration (often prefilled by admin)
326	extra_user_info = self.manifest.get("extra_user_info", [])	1✔
327	needs_more_info = any(	1✔
328	item["id"] not in user.extra_fields.keys() for item in extra_user_info
329	)
330
331	return len(form) > 0 or needs_more_info	1✔
332
333	def revoke_user(self, user):	1✔
334	if self.skip_user_provisioning:	1✔
335	# should never be triggered
336	return False, "Cannot revoke manual integration"	1✔
337
338	self.new_hire = user	1✔
339	self.has_user_context = True	1✔
340
341	# Renew token if necessary
342	if not self.renew_key():	1✔
343	return False, "Couldn't renew key"	×
344
345	revoke_manifest = self.manifest.get("revoke", [])	1✔
346
347	# add extra fields directly to params
348	self.params = self.new_hire.extra_fields	1✔
349
350	for item in revoke_manifest:	1✔
351	success, response = self.run_request(item)	1✔
352
353	if not success:	1✔
354	return False, self.clean_response(response)	1✔
355
356	return True, ""	1✔
357
358	def renew_key(self):	1✔
359	# Oauth2 refreshing access token if needed
360	success = True	1✔
361	if (	1✔
362	self.has_oauth
363	and "expires_in" in self.extra_args.get("oauth", {})
364	and self.expiring < timezone.now()
365	):
366	success, response = self.run_request(self.manifest["oauth"]["refresh"])	1✔
367
368	if not success:	1✔
369	user = self.new_hire if self.has_user_context else None	1✔
370	Notification.objects.create(	1✔
371	notification_type=Notification.Type.FAILED_INTEGRATION,
372	extra_text=self.name,
373	created_for=user,
374	description="Refresh url: " + str(response),
375	)
376	return success	1✔
377
378	self.extra_args["oauth"] \|= response.json()	1✔
379	if "expires_in" in response.json():	1✔
380	self.expiring = timezone.now() + timedelta(	1✔
381	seconds=response.json()["expires_in"]
382	)
383	self.save(update_fields=["expiring", "extra_args"])	1✔
384	return success	1✔
385
386	def _check_condition(self, response, condition):	1✔
387	value = self._replace_vars(condition.get("value"))	1✔
388	try:	1✔
389	# first argument will be taken from the response
390	response_value = get_value_from_notation(	1✔
391	condition.get("response_notation"), response.json()
392	)
393	except KeyError:	×
394	# we know that the result might not be in the response yet, as we are
395	# waiting for the correct response, so just respond with an empty string
396	response_value = ""	×
397	return value == response_value	1✔
398
399	def _polling(self, item, response):	1✔
400	polling = item.get("polling")	1✔
401	continue_if = item.get("continue_if")	1✔
402	interval = polling.get("interval")	1✔
403	amount = polling.get("amount")	1✔
404
405	got_expected_result = self._check_condition(response, continue_if)	1✔
406	if got_expected_result:	1✔
407	return True, response	×
408
409	tried = 1	1✔
410	while amount > tried:	1✔
411	time.sleep(interval)	1✔
412	success, response = self.run_request(item)	1✔
413	got_expected_result = self._check_condition(response, continue_if)	1✔
414	if got_expected_result:	1✔
415	return True, response	1✔
416	tried += 1	1✔
417	# if exceeding the max amounts, then fail
418	return False, response	1✔
419
420	def execute(self, new_hire=None, params=None):	1✔
421	self.params = params or {}	1✔
422	self.params["responses"] = []	1✔
423	self.params["files"] = {}	1✔
424	self.new_hire = new_hire	1✔
425	self.has_user_context = new_hire is not None	1✔
426
427	if self.has_user_context:	1✔
428	self.params \|= new_hire.extra_fields	1✔
429	self.new_hire = new_hire	1✔
430
431	# Renew token if necessary
432	if not self.renew_key():	1✔
433	return False, None	×
434
435	# Add generated secrets
436	for item in self.manifest.get("initial_data_form", []):	1✔
437	if "name" in item and item["name"] == "generate":	1✔
438	self.extra_args[item["id"]] = get_random_string(length=10)	1✔
439
440	# Run all requests
441	for item in self.manifest["execute"]:	1✔
442	success, response = self.run_request(item)	1✔
443
444	# check if we need to poll before continuing
445	if polling := item.get("polling", False):	1✔
446	success, response = self._polling(item, response)	1✔
447
448	# check if we need to block this integration based on condition
449	if continue_if := item.get("continue_if", False):	1✔
450	got_expected_result = self._check_condition(response, continue_if)	1✔
451	if not got_expected_result:	1✔
452	response = self.clean_response(response=response)	1✔
453	Notification.objects.create(	1✔
454	notification_type=Notification.Type.BLOCKED_INTEGRATION,
455	extra_text=self.name,
456	created_for=new_hire,
457	description=f"Execute url ({item['url']}): {response}",
458	)
459	return False, response	1✔
460
461	# No need to retry or log when we are importing users
462	if not success:	1✔
463	if self.has_user_context:	1✔
464	response = self.clean_response(response=response)	1✔
465	if polling:	1✔
466	response = "Polling timed out: " + response	×
467	Notification.objects.create(	1✔
468	notification_type=Notification.Type.FAILED_INTEGRATION,
469	extra_text=self.name,
470	created_for=new_hire,
471	description=f"Execute url ({item['url']}): {response}",
472	)
473	# Retry url in one hour
474	try:	1✔
475	schedule(	1✔
476	"admin.integrations.tasks.retry_integration",
477	new_hire.id,
478	self.id,
479	params,
480	name=(
481	f"Retrying integration {self.id} for new hire {new_hire.id}"
482	),
483	next_run=timezone.now() + timedelta(hours=1),
484	schedule_type=Schedule.ONCE,
485	)
486	except: # noqa E722	×
487	# Only errors when item gets added another time, so we can safely
488	# let it pass.
489	pass	×
490	return False, response	1✔
491
492	# save if file, so we can reuse later
493	save_as_file = item.get("save_as_file")	1✔
494	if save_as_file is not None:	1✔
495	self.params["files"][save_as_file] = io.BytesIO(response.content)	1✔
496
497	# save json response temporarily to be reused in other parts
498	try:	1✔
499	self.params["responses"].append(response.json())	1✔
500	except: # noqa E722	×
501	# if we save a file, then just append an empty dict
502	self.params["responses"].append({})	×
503
504	# store data coming back from response to the user, so we can reuse in other
505	# integrations
506	if store_data := item.get("store_data", {}):	1✔
507	for new_hire_prop, notation_for_response in store_data.items():	1✔
508	try:	1✔
509	value = get_value_from_notation(	1✔
510	notation_for_response, response.json()
511	)
512	except KeyError:	1✔
513	return (	1✔
514	False,
515	f"Could not store data to new hire: {notation_for_response}"
516	f" not found in {self.clean_response(response.json())}",
517	)
518
519	# save to new hire and to temp var `params` on this model for use in
520	# the same integration
521	new_hire.extra_fields[new_hire_prop] = value	1✔
522	self.params[new_hire_prop] = value	1✔
523	new_hire.save()	1✔
524
525	# Run all post requests (notifications)
526	for item in self.manifest.get("post_execute_notification", []):	1✔
527	if item["type"] == "email":	1✔
528	send_email_with_notification(	1✔
529	subject=self._replace_vars(item["subject"]),
530	message=self._replace_vars(item["message"]),
531	to=self._replace_vars(item["to"]),
532	notification_type=(
533	Notification.Type.SENT_EMAIL_INTEGRATION_NOTIFICATION
534	),
535	)
536	return True, None	1✔
537	else:
538	try:	×
539	client = Client(	×
540	settings.TWILIO_ACCOUNT_SID, settings.TWILIO_AUTH_TOKEN
541	)
542	client.messages.create(	×
543	to=new_hire.phone,
544	from_=settings.TWILIO_FROM_NUMBER,
545	body=self._replace_vars(item["message"]),
546	)
547	except Exception:	×
548	Notification.objects.create(	×
549	notification_type=(
550	Notification.Type.FAILED_TEXT_INTEGRATION_NOTIFICATION
551	),
552	extra_text=self.name,
553	created_for=new_hire,
554	)
555	return True, None	×
556
557	# Succesfully ran integration, add notification only when we are provisioning
558	# access
559	if self.has_user_context:	1✔
560	Notification.objects.create(	1✔
561	notification_type=Notification.Type.RAN_INTEGRATION,
562	extra_text=self.name,
563	created_for=new_hire,
564	)
565	return True, response	1✔
566
567	def config_form(self, data=None):	1✔
568	if self.skip_user_provisioning:	1✔
569	from .forms import ManualIntegrationConfigForm	×
570
571	return ManualIntegrationConfigForm(data=data)	×
572
573	else:
574	from .forms import IntegrationConfigForm	1✔
575
576	return IntegrationConfigForm(instance=self, data=data)	1✔
577
578	def clean_response(self, response):	1✔
579	# if json, then convert to string to make it easier to replace values
580	response = str(response)	1✔
581	for name, value in self.extra_args.items():	1✔
582	response = response.replace(	1✔
583	str(value), _("*Secret value for %(name)s*") % {"name": name}
584	)
585
586	return response	1✔
587
588	objects = IntegrationManager()	1✔
589
590
591	@receiver(post_delete, sender=Integration)	1✔
592	def delete_schedule(sender, instance, **kwargs):	1✔
593	Schedule.objects.filter(name=instance.schedule_name).delete()	1✔

chiefonboarding / ChiefOnboarding / 6700616467

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous