scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 154

Add OCI Signer Authentication. Closes #16048, Closes #15654 (#16064)

Browse source 
* Add OCI Signer Authentication. Closes #16048,  Closes #15654

* Fix linting error

* Remove Recommended, Catch None, Trim Whitespace

* Make method clear re Proxy vs SDK
This commit is contained in:
John Lathouwers 2025-10-31 02:59:01 +00:00 • committed by GitHub
parent 0d84c11bbe
commit a6f740f28b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 645 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

199
docs/my-website/docs/providers/oci.md
View file

@ -29,17 +29,38 @@ Check the [OCI Models List](https://docs.oracle.com/en-us/iaas/Content/generativ
## Authentication
LiteLLM uses OCI signing key authentication. Follow the [official Oracle tutorial](https://docs.oracle.com/en-us/iaas/Content/API/Concepts/apisigningkey.htm) to create a signing key and obtain the following parameters:
LiteLLM supports two authentication methods for OCI:
### Method 1: Manual Credentials
Provide individual OCI credentials directly to LiteLLM. Follow the [official Oracle tutorial](https://docs.oracle.com/en-us/iaas/Content/API/Concepts/apisigningkey.htm) to create a signing key and obtain the following parameters:
- `user`
- `fingerprint`
- `tenancy`
- `region`
- `key_file`
- `key_file` or `key`
- `compartment_id`
This is the default method for LiteLLM AI Gateway (LLM Proxy) access to OCI GenAI models.
### Method 2: OCI SDK Signer
Use an OCI SDK `Signer` object for authentication. This method:
- Leverages the official [OCI SDK for signing](https://docs.oracle.com/en-us/iaas/tools/python/latest/api/signing.html)
- Supports additional authentication methods (instance principals, workload identity, etc.)
To use this method, install the OCI SDK:
```bash
pip install oci
```
This method is an alternative when using the LiteLLM SDK on Oracle Cloud Infrastructure (instances or Oracle Kubernetes Engine).
## Usage
Input the parameters obtained from the OCI signing key creation process into the `completion` function.
<Tabs>
<TabItem value="manual" label="Manual Credentials">
Input the parameters obtained from the OCI signing key creation process into the `completion` function:
```python
import os
@ -64,10 +85,119 @@ response = completion(
print(response)
```
</TabItem>
<TabItem value="oci-sdk" label="OCI SDK Signer" default>
Use the OCI SDK `Signer` for authentication:
```python
from litellm import completion
from oci.signer import Signer
# Create an OCI Signer
signer = Signer(
tenancy="ocid1.tenancy.oc1..",
user="ocid1.user.oc1..",
fingerprint="xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx",
private_key_file_location="~/.oci/key.pem",
# Or use private_key_content="<your_private_key_content>"
)
messages = [{"role": "user", "content": "Hey! how's it going?"}]
response = completion(
model="oci/xai.grok-4",
messages=messages,
oci_signer=signer,
oci_region="us-chicago-1", # Optional, defaults to us-ashburn-1
oci_serving_mode="ON_DEMAND", # Optional, default is "ON_DEMAND". Other option is "DEDICATED"
oci_compartment_id="<oci_compartment_id>",
)
print(response)
```
**Alternative: Use OCI Config File**
The OCI SDK can automatically load credentials from `~/.oci/config`:
```python
from litellm import completion
from oci.config import from_file
from oci.signer import Signer
# Load config from file
config = from_file("~/.oci/config", "DEFAULT") # "DEFAULT" is the profile name
signer = Signer(
tenancy=config["tenancy"],
user=config["user"],
fingerprint=config["fingerprint"],
private_key_file_location=config["key_file"],
pass_phrase=config.get("pass_phrase") # Optional if key is encrypted
)
messages = [{"role": "user", "content": "Hey! how's it going?"}]
response = completion(
model="oci/xai.grok-4",
messages=messages,
oci_signer=signer,
oci_region=config["region"],
oci_compartment_id="<oci_compartment_id>",
)
print(response)
```
**Instance Principal Authentication**
For applications running on OCI compute instances:
```python
from litellm import completion
from oci.auth.signers import InstancePrincipalsSecurityTokenSigner
oci.auth.signers.get_oke_workload_identity_resource_principal_signer()
# Use instance principal authentication
signer = InstancePrincipalsSecurityTokenSigner()
messages = [{"role": "user", "content": "Hey! how's it going?"}]
response = completion(
model="oci/xai.grok-4",
messages=messages,
oci_signer=signer,
oci_region="us-chicago-1",
oci_compartment_id="<oci_compartment_id>",
)
print(response)
```
**Use workload identity authentication**
For applications running in Oracle Kubernetes Engine (OKE):
```python
from litellm import completion
from oci.auth.signers import get_oke_workload_identity_resource_principal_signer
# Use instance principal authentication
signer = get_oke_workload_identity_resource_principal_signer()
messages = [{"role": "user", "content": "Hey! how's it going?"}]
response = completion(
model="oci/xai.grok-4",
messages=messages,
oci_signer=signer,
oci_region="us-chicago-1",
oci_compartment_id="<oci_compartment_id>",
)
print(response)
```
</TabItem>
</Tabs>
## Usage - Streaming
Just set `stream=True` when calling completion.
<Tabs>
<TabItem value="manual-stream" label="Manual Credentials">
```python
import os
from litellm import completion
@ -93,10 +223,68 @@ for chunk in response:
print(chunk["choices"][0]["delta"]["content"]) # same as openai format
```
</TabItem>
<TabItem value="oci-sdk-stream" label="OCI SDK Signer" default>
```python
from litellm import completion
from oci.signer import Signer
signer = Signer(
tenancy="ocid1.tenancy.oc1..",
user="ocid1.user.oc1..",
fingerprint="xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx",
private_key_file_location="~/.oci/key.pem",
)
messages = [{"role": "user", "content": "Hey! how's it going?"}]
response = completion(
model="oci/xai.grok-4",
messages=messages,
stream=True,
oci_signer=signer,
oci_region="us-chicago-1",
oci_compartment_id="<oci_compartment_id>",
)
for chunk in response:
print(chunk["choices"][0]["delta"]["content"]) # same as openai format
```
</TabItem>
</Tabs>
## Usage Examples by Model Type
### Using Cohere Models
<Tabs>
<TabItem value="cohere-sdk" label="OCI SDK Signer" default>
```python
from litellm import completion
from oci.signer import Signer
signer = Signer(
tenancy="ocid1.tenancy.oc1..",
user="ocid1.user.oc1..",
fingerprint="xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx",
private_key_file_location="~/.oci/key.pem",
)
messages = [{"role": "user", "content": "Explain quantum computing"}]
response = completion(
model="oci/cohere.command-latest",
messages=messages,
oci_signer=signer,
oci_region="us-chicago-1",
oci_compartment_id="<oci_compartment_id>",
)
print(response)
```
</TabItem>
<TabItem value="cohere-manual" label="Manual Credentials">
```python
from litellm import completion
@ -112,4 +300,7 @@ response = completion(
oci_compartment_id=<oci_compartment_id>,
)
print(response)
```
```
</TabItem>
</Tabs>

328
litellm/llms/oci/chat/transformation.py
View file

@ -2,7 +2,8 @@ import base64
import datetime
import hashlib
import json
from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional, Tuple, Union
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional, Protocol, Tuple, Union
from urllib.parse import urlparse
import httpx
@ -62,6 +63,47 @@ else:
LiteLLMLoggingObj = Any
class OCISignerProtocol(Protocol):
"""
Protocol for OCI request signers (e.g., oci.signer.Signer).
This protocol defines the interface expected for OCI SDK signer objects.
Compatible with the OCI Python SDK's Signer class.
See: https://docs.oracle.com/en-us/iaas/tools/python/latest/api/signing.html
"""
def do_request_sign(self, request: Any, *, enforce_content_headers: bool = False) -> None:
"""
Sign an HTTP request by adding authentication headers.
Args:
request: Request object with method, url, headers, body, and path_url attributes
enforce_content_headers: Whether to enforce content-type and content-length headers
"""
...
@dataclass
class OCIRequestWrapper:
"""
Wrapper for HTTP requests compatible with OCI signer interface.
This class wraps request data in a format compatible with OCI SDK signers,
which expect objects with method, url, headers, body, and path_url attributes.
"""
method: str
url: str
headers: dict
body: bytes
@property
def path_url(self) -> str:
"""Returns the path + query string for OCI signing."""
parsed_url = urlparse(self.url)
return parsed_url.path + ("?" + parsed_url.query if parsed_url.query else "")
def sha256_base64(data: bytes) -> str:
digest = hashlib.sha256(data).digest()
return base64.b64encode(digest).decode()
@ -228,29 +270,89 @@ class OCIChatConfig(BaseConfig):
return adapted_params
def sign_request(
def _sign_with_oci_signer(
self,
headers: dict,
optional_params: dict,
request_data: dict,
api_base: str,
api_key: Optional[str] = None,
model: Optional[str] = None,
stream: Optional[bool] = None,
fake_stream: Optional[bool] = None,
) -> Tuple[dict, Optional[bytes]]:
) -> Tuple[dict, bytes]:
"""
Some providers like Bedrock require signing the request. The sign request funtion needs access to `request_data` and `complete_url`
Args:
headers: dict
optional_params: dict
request_data: dict - the request body being sent in http request
api_base: str - the complete url being sent in http request
Returns:
dict - the signed headers
"""
import json
Sign request using OCI SDK Signer object.
Args:
headers: Request headers to be signed
optional_params: Optional parameters including oci_signer
request_data: The request body dict to be sent in HTTP request
api_base: The complete URL for the HTTP request
Returns:
Tuple of (signed_headers, encoded_body)
Raises:
OCIError: If signing fails
ValueError: If HTTP method is unsupported
"""
oci_signer = optional_params.get("oci_signer")
body = json.dumps(request_data).encode("utf-8")
method = str(optional_params.get("method", "POST")).upper()
if method not in ["POST", "GET", "PUT", "DELETE", "PATCH"]:
raise ValueError(f"Unsupported HTTP method: {method}")
prepared_headers = headers.copy()
prepared_headers.setdefault("content-type", "application/json")
prepared_headers.setdefault("content-length", str(len(body)))
request_wrapper = OCIRequestWrapper(
method=method,
url=api_base,
headers=prepared_headers,
body=body
)
if oci_signer is None:
raise ValueError("oci_signer cannot be None when calling _sign_with_oci_signer")
try:
oci_signer.do_request_sign(request_wrapper, enforce_content_headers=True)
except Exception as e:
raise OCIError(
status_code=500,
message=(
f"Failed to sign request with provided oci_signer: {str(e)}. "
"The signer must implement the OCI SDK Signer interface with a "
"do_request_sign(request, enforce_content_headers=True) method. "
"See: https://docs.oracle.com/en-us/iaas/tools/python/latest/api/signing.html"
)
) from e
headers.update(request_wrapper.headers)
return headers, body
def _sign_with_manual_credentials(
self,
headers: dict,
optional_params: dict,
request_data: dict,
api_base: str,
) -> Tuple[dict, None]:
"""
Sign request using manual OCI credentials.
Args:
headers: Request headers to be signed
optional_params: Optional parameters including OCI credentials
request_data: The request body dict to be sent in HTTP request
api_base: The complete URL for the HTTP request
Returns:
Tuple of (signed_headers, None)
Raises:
Exception: If required credentials are missing
ImportError: If cryptography package is not installed
"""
oci_region = optional_params.get("oci_region", "us-ashburn-1")
api_base = (
api_base
@ -355,6 +457,69 @@ class OCIChatConfig(BaseConfig):
return headers, None
def sign_request(
self,
headers: dict,
optional_params: dict,
request_data: dict,
api_base: str,
api_key: Optional[str] = None,
model: Optional[str] = None,
stream: Optional[bool] = None,
fake_stream: Optional[bool] = None,
) -> Tuple[dict, Optional[bytes]]:
"""
Sign the OCI request by adding authentication headers.
Supports two signing modes:
1. OCI SDK Signer: Use an oci_signer object to sign the request
2. Manual Signing: Use OCI credentials to manually sign the request
Args:
headers: Request headers to be signed
optional_params: Optional parameters including auth credentials or oci_signer
request_data: The request body dict to be sent in HTTP request
api_base: The complete URL for the HTTP request
api_key: Optional API key (not used for OCI)
model: Optional model name
stream: Optional streaming flag
fake_stream: Optional fake streaming flag
Returns:
Tuple of (signed_headers, encoded_body):
- If oci_signer is provided: Returns (headers, body) where body is the encoded JSON
- If manual credentials are provided: Returns (headers, None) as body is not returned
for the manual signing path
Raises:
OCIError: If signing fails with oci_signer
Exception: If required credentials are missing
ImportError: If cryptography package is not installed (manual signing only)
Example:
>>> from oci.signer import Signer
>>> signer = Signer(
... tenancy="ocid1.tenancy.oc1..",
... user="ocid1.user.oc1..",
... fingerprint="xx:xx:xx",
... private_key_file_location="~/.oci/key.pem"
... )
>>> headers, body = config.sign_request(
... headers={},
... optional_params={"oci_signer": signer},
... request_data={"message": "Hello"},
... api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/..."
... )
"""
oci_signer = optional_params.get("oci_signer")
# If a signer is provided, use it for request signing
if oci_signer is not None:
return self._sign_with_oci_signer(headers, optional_params, request_data, api_base)
# Standard manual credential signing
return self._sign_with_manual_credentials(headers, optional_params, request_data, api_base)
def validate_environment(
self,
headers: dict,
@ -365,36 +530,67 @@ class OCIChatConfig(BaseConfig):
api_key: Optional[str] = None,
api_base: Optional[str] = None,
) -> dict:
"""
Validate the OCI environment and credentials.
Supports two authentication modes:
1. OCI SDK Signer: Pass an oci_signer object (e.g., oci.signer.Signer)
2. Manual Credentials: Pass oci_user, oci_fingerprint, oci_tenancy, and oci_key/oci_key_file
Args:
headers: Request headers to populate
model: Model name
messages: List of chat messages
optional_params: Optional parameters including authentication credentials
litellm_params: LiteLLM parameters
api_key: Optional API key (not used for OCI)
api_base: Optional API base URL
Returns:
Updated headers dict
Raises:
Exception: If required parameters are missing or invalid
"""
oci_signer = optional_params.get("oci_signer")
oci_region = optional_params.get("oci_region", "us-ashburn-1")
# Determine api_base
api_base = (
api_base
or litellm.api_base
or f"https://inference.generativeai.{oci_region}.oci.oraclecloud.com"
)
oci_user = optional_params.get("oci_user")
oci_fingerprint = optional_params.get("oci_fingerprint")
oci_tenancy = optional_params.get("oci_tenancy")
oci_key = optional_params.get("oci_key")
oci_key_file = optional_params.get("oci_key_file")
oci_compartment_id = optional_params.get("oci_compartment_id")
if (
not oci_user
or not oci_fingerprint
or not oci_tenancy
or not (oci_key or oci_key_file)
or not oci_compartment_id
):
raise Exception(
"Missing required parameters: oci_user, oci_fingerprint, oci_tenancy, oci_compartment_id "
"and at least one of oci_key or oci_key_file."
)
if not api_base:
raise Exception(
"Either `api_base` must be provided or `litellm.api_base` must be set. Alternatively, you can set the `oci_region` optional parameter to use the default OCI region."
"Either `api_base` must be provided or `litellm.api_base` must be set. "
"Alternatively, you can set the `oci_region` optional parameter to use the default OCI region."
)
# Validate credentials only if signer is not provided
if oci_signer is None:
oci_user = optional_params.get("oci_user")
oci_fingerprint = optional_params.get("oci_fingerprint")
oci_tenancy = optional_params.get("oci_tenancy")
oci_key = optional_params.get("oci_key")
oci_key_file = optional_params.get("oci_key_file")
oci_compartment_id = optional_params.get("oci_compartment_id")
if (
not oci_user
or not oci_fingerprint
or not oci_tenancy
or not (oci_key or oci_key_file)
or not oci_compartment_id
):
raise Exception(
"Missing required parameters: oci_user, oci_fingerprint, oci_tenancy, oci_compartment_id "
"and at least one of oci_key or oci_key_file. "
"Alternatively, provide an oci_signer object from the OCI SDK."
)
# Common header setup
headers.update(
{
"content-type": "application/json",
@ -442,12 +638,12 @@ class OCIChatConfig(BaseConfig):
for openai_key, oci_key in open_ai_to_oci_param_map.items():
if oci_key and openai_key in optional_params:
selected_params[oci_key] = optional_params[openai_key] # type: ignore[index]
# Also check for already-mapped OCI params (for backward compatibility)
for oci_value in open_ai_to_oci_param_map.values():
if oci_value and oci_value in optional_params and oci_value not in selected_params:
selected_params[oci_value] = optional_params[oci_value] # type: ignore[index]
if "tools" in selected_params:
if vendor == OCIVendors.COHERE:
selected_params["tools"] = self.adapt_tool_definitions_to_cohere_standard( # type: ignore[assignment]
@ -465,7 +661,7 @@ class OCIChatConfig(BaseConfig):
for msg in messages[:-1]: # All messages except the last one
role = msg.get("role")
content = msg.get("content")
if isinstance(content, list):
# Extract text from content array
text_content = ""
@ -473,11 +669,11 @@ class OCIChatConfig(BaseConfig):
if isinstance(content_item, dict) and content_item.get("type") == "text":
text_content += content_item.get("text", "")
content = text_content
# Ensure content is a string
if not isinstance(content, str):
content = str(content) if content is not None else ""
# Handle tool calls
tool_calls: Optional[List[CohereToolCall]] = None
if role == "assistant" and "tool_calls" in msg and msg.get("tool_calls"): # type: ignore[union-attr,typeddict-item]
@ -492,12 +688,12 @@ class OCIChatConfig(BaseConfig):
arguments = {}
else:
arguments = raw_arguments
tool_calls.append(CohereToolCall(
name=str(tool_call.get("function", {}).get("name", "")),
parameters=arguments
))
if role == "user":
chat_history.append(CohereMessage(role="USER", message=content))
elif role == "assistant":
@ -505,11 +701,11 @@ class OCIChatConfig(BaseConfig):
elif role == "tool":
# Tool messages need special handling
chat_history.append(CohereMessage(
role="TOOL",
role="TOOL",
message=content,
toolCalls=None # Tool messages don't have tool calls
))
return chat_history
def adapt_tool_definitions_to_cohere_standard(self, tools: List[Dict[str, Any]]) -> List[CohereTool]:
@ -519,7 +715,7 @@ class OCIChatConfig(BaseConfig):
function_def = tool.get("function", {})
parameters = function_def.get("parameters", {}).get("properties", {})
required = function_def.get("parameters", {}).get("required", [])
parameter_definitions = {}
for param_name, param_schema in parameters.items():
parameter_definitions[param_name] = CohereParameterDefinition(
@ -527,13 +723,13 @@ class OCIChatConfig(BaseConfig):
type=param_schema.get("type", "string"),
isRequired=param_name in required
)
cohere_tools.append(CohereTool(
name=function_def.get("name", ""),
description=function_def.get("description", ""),
parameterDefinitions=parameter_definitions
))
return cohere_tools
def _extract_text_content(self, content: Any) -> str:
@ -586,7 +782,7 @@ class OCIChatConfig(BaseConfig):
user_messages = [msg for msg in messages if msg.get("role") == "user"]
if not user_messages:
raise Exception("No user message found for Cohere model")
# Create Cohere-specific chat request
chat_request = CohereChatRequest(
@ -595,7 +791,7 @@ class OCIChatConfig(BaseConfig):
chatHistory=self.adapt_messages_to_cohere_standard(messages),
**self._get_optional_params(OCIVendors.COHERE, optional_params)
)
data = OCICompletionPayload(
compartmentId=oci_compartment_id,
servingMode=servingMode,
@ -616,24 +812,24 @@ class OCIChatConfig(BaseConfig):
return data.model_dump(exclude_none=True)
def _handle_cohere_response(
self,
json_response: dict,
model: str,
self,
json_response: dict,
model: str,
model_response: ModelResponse
) -> ModelResponse:
"""Handle Cohere-specific response format."""
cohere_response = CohereChatResult(**json_response)
# Cohere response format (uses camelCase)
model_id = model
# Set basic response info
model_response.model = model_id
model_response.created = int(datetime.datetime.now().timestamp())
# Extract the response text
response_text = cohere_response.chatResponse.text
oci_finish_reason = cohere_response.chatResponse.finishReason
# Map finish reason
if oci_finish_reason == "COMPLETE":
finish_reason = "stop"
@ -641,7 +837,7 @@ class OCIChatConfig(BaseConfig):
finish_reason = "length"
else:
finish_reason = "stop"
# Handle tool calls
tool_calls: Optional[List[Dict[str, Any]]] = None
if cohere_response.chatResponse.toolCalls:
@ -655,7 +851,7 @@ class OCIChatConfig(BaseConfig):
"arguments": json.dumps(tool_call.parameters)
}
})
# Create choice
from litellm.types.utils import Choices
choice = Choices(
@ -668,7 +864,7 @@ class OCIChatConfig(BaseConfig):
finish_reason=finish_reason
)
model_response.choices = [choice]
# Extract usage info
usage_info = cohere_response.chatResponse.usage
from litellm.types.utils import Usage
@ -677,13 +873,13 @@ class OCIChatConfig(BaseConfig):
completion_tokens=usage_info.completionTokens, # type: ignore[union-attr]
total_tokens=usage_info.totalTokens # type: ignore[union-attr]
)
return model_response
def _handle_generic_response(
self,
json: dict,
model: str,
self,
json: dict,
model: str,
model_response: ModelResponse,
raw_response: httpx.Response
) -> ModelResponse:
@ -695,7 +891,7 @@ class OCIChatConfig(BaseConfig):
message=f"Response cannot be casted to OCICompletionResponse: {str(e)}",
status_code=raw_response.status_code,
)
iso_str = completion_response.chatResponse.timeCreated
dt = datetime.datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
model_response.created = int(dt.timestamp())
@ -751,7 +947,7 @@ class OCIChatConfig(BaseConfig):
)
vendor = get_vendor_from_model(model)
# Handle response based on vendor type
if vendor == OCIVendors.COHERE:
model_response = self._handle_cohere_response(json, model, model_response)
@ -1080,7 +1276,7 @@ class OCIStreamWrapper(CustomStreamWrapper):
if not chunk.startswith("data:"):
raise ValueError(f"Chunk does not start with 'data:': {chunk}")
dict_chunk = json.loads(chunk[5:]) # Remove 'data: ' prefix and parse JSON
# Check if this is a Cohere stream chunk
if "apiFormat" in dict_chunk and dict_chunk.get("apiFormat") == "COHERE":
return self._handle_cohere_stream_chunk(dict_chunk)

189
tests/test_litellm/llms/oci/chat/test_oci_chat_transformation.py
View file

@ -11,7 +11,7 @@ import litellm
sys.path.insert(0, os.path.abspath("../../../../.."))
from litellm import ModelResponse
from litellm.llms.oci.chat.transformation import OCIChatConfig, version
from litellm.llms.oci.chat.transformation import OCIChatConfig, OCIRequestWrapper, version
TEST_MODEL_NAME = "xai.grok-4"
TEST_MODEL = f"oci/{TEST_MODEL_NAME}"
@ -309,3 +309,190 @@ class TestOCIChatConfig:
assert usage.prompt_tokens == 10 # type: ignore
assert usage.completion_tokens == 20 # type: ignore
assert usage.total_tokens == 30 # type: ignore
class TestOCISignerSupport:
"""Tests for OCI SDK Signer integration."""
def test_validate_environment_with_oci_signer(self):
"""Test validation when using oci_signer instead of manual credentials."""
config = OCIChatConfig()
headers = {}
# Mock signer object
class MockSigner:
def do_request_sign(self, request, enforce_content_headers=True):
request.headers["authorization"] = "Signature version=\"1\""
optional_params = {
"oci_signer": MockSigner(),
"oci_region": "us-ashburn-1"
}
result = config.validate_environment(
headers=headers,
model=TEST_MODEL,
messages=TEST_MESSAGES, # type: ignore
optional_params=optional_params,
litellm_params={},
)
assert result["content-type"] == "application/json"
assert result["user-agent"] == f"litellm/{version}"
def test_validate_environment_with_oci_signer_no_compartment_id_in_validate(self):
"""Test that oci_compartment_id is not required in validate_environment when using signer."""
config = OCIChatConfig()
headers = {}
class MockSigner:
def do_request_sign(self, request, enforce_content_headers=True):
request.headers["authorization"] = "Signature version=\"1\""
optional_params = {
"oci_signer": MockSigner(),
"oci_region": "us-phoenix-1"
}
# Should not raise an exception even without oci_compartment_id
result = config.validate_environment(
headers=headers,
model=TEST_MODEL,
messages=TEST_MESSAGES, # type: ignore
optional_params=optional_params,
litellm_params={},
)
assert result["content-type"] == "application/json"
def test_sign_request_with_oci_signer(self):
"""Test request signing with oci_signer."""
config = OCIChatConfig()
class MockSigner:
def do_request_sign(self, request, enforce_content_headers=True):
request.headers["authorization"] = "Signature version=\"1\""
request.headers["date"] = "Mon, 01 Jan 2024 00:00:00 GMT"
optional_params = {
"oci_signer": MockSigner(),
"method": "POST"
}
headers, body = config.sign_request(
headers={},
optional_params=optional_params,
request_data={"test": "data"},
api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/20231130/actions/chat"
)
assert "authorization" in headers
assert "Signature" in headers["authorization"]
assert body is not None # oci_signer path returns body
assert json.loads(body.decode("utf-8")) == {"test": "data"}
def test_sign_request_with_oci_signer_updates_headers(self):
"""Test that signer properly updates request headers."""
config = OCIChatConfig()
class MockSigner:
def do_request_sign(self, request, enforce_content_headers=True):
# Verify the request has the expected attributes
assert hasattr(request, "method")
assert hasattr(request, "url")
assert hasattr(request, "headers")
assert hasattr(request, "body")
assert hasattr(request, "path_url")
# Add signature headers
request.headers["authorization"] = "Signature keyId=\"test\""
request.headers["date"] = "Mon, 01 Jan 2024 00:00:00 GMT"
request.headers["x-content-sha256"] = "test-hash"
optional_params = {
"oci_signer": MockSigner(),
}
headers, body = config.sign_request(
headers={"custom-header": "custom-value"},
optional_params=optional_params,
request_data={"message": "Hello"},
api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/20231130/actions/chat"
)
# Check that all signer-added headers are present
assert headers["authorization"] == "Signature keyId=\"test\""
assert headers["date"] == "Mon, 01 Jan 2024 00:00:00 GMT"
assert headers["x-content-sha256"] == "test-hash"
# Original headers should be preserved
assert headers["custom-header"] == "custom-value"
def test_sign_request_with_failing_oci_signer(self):
"""Test error handling when oci_signer fails."""
config = OCIChatConfig()
class FailingSigner:
def do_request_sign(self, request, enforce_content_headers=True):
raise RuntimeError("Signing failed due to invalid credentials")
optional_params = {
"oci_signer": FailingSigner(),
}
from litellm.llms.oci.common_utils import OCIError
with pytest.raises(OCIError) as excinfo:
config.sign_request(
headers={},
optional_params=optional_params,
request_data={"test": "data"},
api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/20231130/actions/chat"
)
assert "Failed to sign request with provided oci_signer" in str(excinfo.value)
assert excinfo.value.status_code == 500
def test_sign_request_with_invalid_http_method(self):
"""Test that invalid HTTP methods are rejected."""
config = OCIChatConfig()
class MockSigner:
def do_request_sign(self, request, enforce_content_headers=True):
pass
optional_params = {
"oci_signer": MockSigner(),
"method": "INVALID"
}
with pytest.raises(ValueError) as excinfo:
config.sign_request(
headers={},
optional_params=optional_params,
request_data={"test": "data"},
api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/20231130/actions/chat"
)
assert "Unsupported HTTP method: INVALID" in str(excinfo.value)
def test_oci_request_wrapper_path_url(self):
"""Test OCIRequestWrapper path_url property."""
wrapper = OCIRequestWrapper(
method="POST",
url="https://example.com/api/v1/chat?param1=value1&param2=value2",
headers={},
body=b"test"
)
assert wrapper.path_url == "/api/v1/chat?param1=value1&param2=value2"
def test_oci_request_wrapper_path_url_no_query(self):
"""Test OCIRequestWrapper path_url property without query string."""
wrapper = OCIRequestWrapper(
method="POST",
url="https://example.com/api/v1/chat",
headers={},
body=b"test"
)
assert wrapper.path_url == "/api/v1/chat"