request_maker.py

# Copyright (c) 2025 Splunk Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import Optional

import requests
from bs4 import UnicodeDammit
from soar_sdk.abstract import SOARClient
from soar_sdk.action_results import ActionOutput
from soar_sdk.exceptions import ActionFailure

from . import helpers
from .asset import Asset
from .auth import OAuth
from .common import logger
from .auth import Authorization, CertificateAuth
from .helpers import temp_cert_files


def make_request(
    asset: Asset,
    soar: SOARClient,
    method: str,
    location: str,
    output: type[ActionOutput],
    verify: bool,
    headers: Optional[str],
    body: Optional[str],
) -> ActionOutput:
    """
    The central engine for making all HTTP requests.

    This function handles parameter parsing, URL construction, authentication,
    request execution (with OAuth token refresh logic), and response processing.
    """

    logger.info(f"Preparing to make {method} http request.")
    parsed_headers = helpers.parse_headers(headers)

    full_url = asset.base_url.rstrip("/") + "/" + location.lstrip("/")

    logger.info(f"Making {method} request to: {full_url}")

    body = UnicodeDammit(body).unicode_markup.encode("utf-8") if isinstance(body, str) else body

    from .app import get_auth_method

    auth_method = get_auth_method(asset, soar)

    if isinstance(auth_method, CertificateAuth):
        return _execute_certificate_request(auth_method, full_url, method, body, verify, parsed_headers, output, asset)
    else:
        return _execute_standard_request(auth_method, full_url, method, body, verify, parsed_headers, output, asset)


def _execute_standard_request(
    auth_method: Authorization,
    full_url: str,
    method: str,
    body: Optional[str],
    verify: bool,
    headers: dict,
    output_cls: type[ActionOutput],
    asset: Asset,
) -> ActionOutput:
    """
    Executes a standard HTTP request (non-certificate based).
    Handles OAuth token refresh logic.
    """
    retries = 1
    response = None

    while retries >= 0:
        auth_object, final_headers = auth_method.create_auth(headers.copy())

        try:
            response = requests.request(
                method=method,
                url=full_url,
                auth=auth_object,
                data=body,
                verify=verify,
                headers=final_headers,
                timeout=asset.timeout,
            )
            response.raise_for_status()
            break

        except requests.exceptions.RequestException as e:
            if isinstance(auth_method, OAuth) and retries > 0:
                logger.warning("Request failed with 401, token might be expired. Forcing a refresh.")
                auth_method.get_token(force_new=True, full_url=full_url)
                retries -= 1
                continue
            else:
                raise ActionFailure(f"Request failed for {full_url}. Details: {e}") from e

    if response is None:
        raise ActionFailure(f"Request failed for {full_url} and no response was received after retries.")

    return response_maker(response, full_url, method, output_cls)


def _execute_certificate_request(
    auth_method: CertificateAuth,
    full_url: str,
    method: str,
    body: Optional[str],
    verify: bool,
    headers: dict,
    output_cls: type[ActionOutput],
    asset: Asset,
) -> ActionOutput:
    """
    Executes an HTTP request using client-side certificates.
    """
    public_cert_data, private_key_data = auth_method.create_auth(headers.copy())

    with temp_cert_files(public_cert_data, private_key_data) as cert_param:
        try:
            response = requests.request(
                method=method,
                url=full_url,
                cert=cert_param,
                data=body,
                verify=verify,
                headers=headers,
                timeout=asset.timeout,
            )
            response.raise_for_status()

        except requests.exceptions.RequestException as e:
            raise ActionFailure(f"Certificate-based request failed for {full_url}. Details: {e}") from e

    return response_maker(response, full_url, method, output_cls)


def response_maker(response, full_url, method, output_cls):
    parsed_body, raw_body = helpers.handle_various_response(response)
    logger.info(f"Successfully processed data. Status: {response.status_code}")

    return output_cls(
        status_code=response.status_code,
        location=full_url,
        method=method,
        parsed_response_body=parsed_body,
        response_body=raw_body,
        response_headers=str(dict(response.headers)),
    )