suite/website_sale_signifyd/models/signifyd_connector.py

# Part of Hibou Suite Professional. See LICENSE_PROFESSIONAL file for full copyright and licensing details.

import requests
from datetime import datetime as dt
from base64 import b64encode, b64decode
import json

import hmac
import hashlib

from odoo import api, fields, models
from .signifyd_api import SignifydAPI


class SignifydConnector(models.Model):
    _name = 'signifyd.connector'
    _description = 'Interact with Signifyd API'

    name = fields.Char(string='Connector Name', required=True)
    secret_key = fields.Char(string='API Key', required=True)
    webhooks_registered = fields.Boolean(string='Successfully Registered Webhooks')
    notify_user_ids = fields.Many2many('res.users', string='Receive decline notifications')
    website_ids = fields.One2many('website', 'signifyd_connector_id', string='Used on Websites')
    # TODO: remove options no longer available in api v3
    # signifyd_case_type = fields.Selection([
    #     ('', 'No Case'),
    #     ('SCORE', 'Score'),
    #     ('DECISION', 'Decision'),
    #     ('GUARANTEE', 'Guarantee'),
    # ], string='Default Case Creation', help='Used for internal/admin orders, overridden by payment acquirer.',
    #     required=True, default='')
    signifyd_coverage_ids = fields.Many2many('signifyd.coverage', string='Available Coverage Types',
        help='Note that exclusive coverage types will only allow one to be selected.')
    teamid = fields.Char(string='Signifyd Team ID')

    @api.onchange('signifyd_coverage_ids')
    def _onchange_signifyd_coverage_ids(self):
        self.signifyd_coverage_ids = self.signifyd_coverage_ids._apply_exclusivity()


    def get_connection(self):
        if not self:
            return
        return SignifydAPI(self.secret_key, self.teamid)

    def register_webhooks(self):
        self.ensure_one()
        # we may need a better way to link the connector to the website.
        base_url = None
        website = self.env['website'].search([('signifyd_connector_id', '=', self.id)], limit=1)
        if website and website.domain:
            base_url = website.domain
            if base_url.find('://') <= 0:  # documentation says if no protocol use http
                base_url = 'http://' + base_url
        if not base_url:
            base_url = self.env['ir.config_parameter'].sudo().get_param('web.base.url')
        api = self.get_connection()
        r = api.register_webhook(base_url + '/signifyd/cases/update')
        r.raise_for_status()
        return r

    def action_register_webhooks(self):
        notification = {
            'type': 'ir.actions.client',
            'tag': 'display_notification',
            'params': {
                'title': ('Signifyd Connector'),
                'sticky': True,
            },
        }

        res = self.register_webhooks()

        if 200 <= res.status_code < 300:
            notification['params']['type'] = 'success'
            notification['params']['message'] = 'Successfully registered webhooks with Signifyd.'
            self.webhooks_registered = True
            return notification

        else:
            notification['params']['type'] = 'danger'
            notification['params']['message'] = res.content.decode('utf-8')
            try:
                # trying to make a better error, not be exhaustive with error handling.
                object = json.loads(notification['params']['message'])
                notification['params']['message'] = '\n'.join([e[0] for e in (object.get('errors') or {}).values()])
            except:
                pass
            self.webhooks_registered = False
            return notification

    def _check_webhook_signature(self, request):
        signature = b64decode(request.httprequest.headers.get('SIGNIFYD-SEC-HMAC-SHA256'))
        if not signature:
            raise Exception('Signifyd webhook missing signature')

        key = self.secret_key.encode()
        digest = hmac.new(key, request.httprequest.data, hashlib.sha256).digest()
        if signature != digest:
            raise Exception('Signifyd webhook signature does not match')

    def process_post_values(self, post):
        # Construct dict from request data for endpoints
        uuid = post.get('uuid')
        case_id = post.get('caseId')
        team_name = post.get('teamName')
        team_id = post.get('teamId')
        review_disposition = post.get('reviewDisposition')
        guarantee_disposition = post.get('guaranteeDisposition')
        order_outcome = post.get('orderOutcome')
        status = post.get('status')
        score = post.get('score')
        disposition_reason = post.get('dispositionReason')
        disposition = post.get('disposition')
        checkpoint_action = post.get('checkpointAction', '')
        if not checkpoint_action and guarantee_disposition:
            if guarantee_disposition == 'APPROVED':
                checkpoint_action = 'ACCEPT'
            elif guarantee_disposition == 'DECLINED':
                checkpoint_action = 'REJECT'
            else:
                checkpoint_action = 'HOLD'
        last_update = str(dt.now())

        values = {}

        # Validate that the order and case match the request
        values.update({'uuid': uuid}) if uuid else ''
        values.update({'team_name': team_name}) if team_name else ''
        values.update({'team_id': team_id}) if team_id else ''
        values.update({'review_disposition': review_disposition}) if review_disposition else ''
        values.update({'guarantee_disposition': guarantee_disposition}) if guarantee_disposition else ''
        values.update({'order_outcome': order_outcome}) if order_outcome else ''
        values.update({'status': status}) if status else ''
        values.update({'score': score}) if score else ''
        values.update({'case_id': case_id}) if case_id else ''
        values.update({'disposition_reason': disposition_reason}) if disposition_reason else ''
        values.update({'disposition': disposition}) if disposition else ''
        values.update({'last_update': last_update})
        values.update({'checkpoint_action': checkpoint_action})

        return values