Outbound callback source reference

Source commit: 796717af517d50d52249a484fa35c42394ada304.

Generated from this revision's source declarations. These defaults and traits do not report a running installation's active settings, prove provider delivery, or establish that newly added components are integrated.

These excerpts are read directly from source with Python's AST; callback runtime modules are not imported. The build provenance hashes each listed module. The verifier, URL construction and captured-attempt acceptance code below are regenerated on every source build.

Maintained Phaxio setup and verification guidance. This reference covers outbound observations; it does not establish inbound verification readiness.

_pairs

Source: api/app/provider_signatures.py.

def _pairs(values, value_type):
    if not isinstance(values, Sequence) or isinstance(values, (str, bytes, bytearray)):
        return None
    pairs = []
    for pair in values:
        if (not isinstance(pair, Sequence) or isinstance(pair, (str, bytes, bytearray))
                or len(pair) != 2 or not isinstance(pair[0], str) or not isinstance(pair[1], value_type)):
            return None
        pairs.append((pair[0], pair[1]))
    # Stable sorting retains wire order for repeated names and never mutates input.
    return sorted(pairs, key=lambda pair: pair[0])

verify_phaxio_signature

Source: api/app/provider_signatures.py.

def verify_phaxio_signature(
    callback_token: str,
    callback_url: str,
    fields: Sequence[tuple[str, str]],
    files: Sequence[tuple[str, bytes]],
    signature: str,
) -> bool:
    """Verify Phaxio's lowercase hex SHA1 HMAC with its distinct Callback Token.

    URL is the exact submitted public URL, including query/trailing slash.
    Ordered repeated fields/file parts are retained; file content uses SHA1 hex.
    Malformed input fails closed. The send API secret is not a substitute token.
    """
    try:
        if (not isinstance(callback_token, str) or not callback_token or not isinstance(signature, str)
                or re.fullmatch(r'[0-9a-f]{40}', signature) is None or _public_url(callback_url) is None):
            return False
        parameters, attachments = _pairs(fields, str), _pairs(files, bytes)
        if parameters is None or attachments is None:
            return False
        message = callback_url + ''.join(name + value for name, value in parameters)
        message += ''.join(name + hashlib.sha1(content).hexdigest() for name, content in attachments)
        expected = hmac.new(callback_token.encode('utf-8'), message.encode('utf-8'), hashlib.sha1).hexdigest()
        return hmac.compare_digest(expected, signature)
    except (TypeError, ValueError):
        return False

callback_base_url

Source: api/app/callback_locator.py.

def callback_base_url(revision, profile):
    """Use the captured override or this installation's captured public URL."""
    configuration = profile.configuration
    explicit = configuration.settings.get('callback_url')
    if explicit:
        return explicit
    return revision.values.public_api_url.rstrip('/') + '/' + configuration.provider_id + '-callback'

callback_url_with_locators

Source: api/app/callback_locator.py.

def callback_url_with_locators(callback_url: str, job_id: str, attempt_id: str | None = None) -> str:
    """Preserve unrelated query bytes/order; replace reserved locators once.

    The caller supplies its captured public URL, never proxy/request headers.
    Omitting attempt_id removes stale attempt locators for the legacy interface.
    """
    try:
        if (not isinstance(callback_url, str) or not callback_url
                or any(ord(char) <= 32 or ord(char) == 127 for char in callback_url)
                or '#' in callback_url):
            raise ValueError
        parsed = urlsplit(callback_url)
        if (parsed.scheme not in {'http', 'https'} or not parsed.hostname
                or parsed.username is not None or parsed.password is not None):
            raise ValueError
        parsed.port
        if (not isinstance(job_id, str) or not job_id
                or (attempt_id is not None and (not isinstance(attempt_id, str) or not attempt_id))):
            raise ValueError
        callback_url.encode('utf-8')
        remaining = [part for part in parsed.query.split('&') if
            unquote_plus(part.partition('=')[0]) not in {'job_id', 'attempt_id'}] if parsed.query else []
        locators = [('job_id', job_id)]
        if attempt_id is not None:
            locators.append(('attempt_id', attempt_id))
        query = '&'.join(remaining + [urlencode(locators)])
        return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, query, ''))
    except (TypeError, ValueError):
        raise ValueError('Invalid public callback URL or locator.') from None

CapturedCallbacks.receive

Source: api/app/outbound_callbacks.py.

def receive(self, provider, job_id, attempt_id, *, fields, files, signature):
    if (provider not in {'phaxio', 'signalwire'}
            or not isinstance(job_id, str) or re.fullmatch('[a-f0-9]{32}', job_id) is None
            or not isinstance(attempt_id, str) or re.fullmatch('[a-f0-9]{32}', attempt_id) is None):
        raise CallbackRejected()
    try:
        row = self.store.get(job_id)
    except DeliveryConflict:
        raise CallbackRejected() from None
    if row['attempt_id'] != attempt_id:
        raise CallbackRejected()
    # Authenticate against the account this attempt used, never the fax's default.
    revision, profile = self.store.attempt_context(job_id, attempt_id)
    configuration = profile.configuration
    if configuration.provider_id != provider or configuration.manifest is not None:
        raise CallbackRejected()
    url = callback_url_with_locators(callback_base_url(revision, profile), job_id, attempt_id)
    credentials = configuration.credentials
    if provider == 'phaxio':
        # Disabled verification means callbacks are disabled; it never grants
        # an unsigned request authority to change a delivery result.
        if configuration.settings.get('verify_signature', True) is not True:
            raise CallbackRejected()
        authenticated = verify_phaxio_signature(credentials.get('callback_token', ''), url, fields, files, signature)
    else:
        authenticated = not files and verify_signalwire_signature(
            credentials.get('webhook_signing_key', ''), url, fields, signature)
    if not authenticated:
        raise CallbackRejected()
    try:
        if provider == 'phaxio':
            sid, status = _phaxio_payload(fields)
        else:
            sid, status = _one(fields, {'FaxSid', 'sid'}), _one(fields, {'FaxStatus', 'status'})
        status = normalize_status(status)
        # Match signature ordering: distinct names may move, but repeated
        # names retain their signed wire order. Never sort their values.
        event = json.dumps({'fields': sorted(fields, key=lambda part: part[0]),
            'files': [(name, hashlib.sha256(data).hexdigest())
                      for name, data in sorted(files, key=lambda part: part[0])]},
            sort_keys=True, separators=(',', ':'))
        key = 'callback:' + hashlib.sha256(event.encode()).hexdigest()
    except (TypeError, ValueError):
        raise CallbackRejected() from None
    return self.store.observe(job_id, attempt_id=attempt_id, profile_id=profile.id,
                              provider_sid=sid, status=status, event_key=key)