diff --git a/src/oci-api-mcp-server/CHANGELOG.md b/src/oci-api-mcp-server/CHANGELOG.md index 881362f8..39a58f44 100644 --- a/src/oci-api-mcp-server/CHANGELOG.md +++ b/src/oci-api-mcp-server/CHANGELOG.md @@ -10,6 +10,13 @@ - OCI command parsing now preserves quoted arguments when retrieving command help or running OCI CLI commands. ([#100](https://github.com/oracle/mcp/issues/100)) +## 2.1.0 - 2026-07-08 + +### Added + +- Added environment-driven JWT-to-UPST authentication for OCI CLI commands, including + automatic token renewal and private, isolated credential storage. + ## 2.0.0 ### Breaking Changes diff --git a/src/oci-api-mcp-server/README.md b/src/oci-api-mcp-server/README.md index 087a7c50..c2d75bcc 100644 --- a/src/oci-api-mcp-server/README.md +++ b/src/oci-api-mcp-server/README.md @@ -27,6 +27,28 @@ uvx oracle.oci-api-mcp-server ⚠️ **NOTE**: All actions use the configured OCI CLI profile. Use least-privilege IAM and protect secrets. +## JWT-to-UPST authentication + +Set the following variables to authenticate OCI CLI commands with an identity-domain +service bearer token exchanged for an OCI User Principal Security Token (UPST): + +```sh +export OCI_UPST_DOMAIN_URL='https://.identity.oraclecloud.com' +export OCI_UPST_CLIENT_ID='' +export OCI_UPST_CLIENT_SECRET='' +export OCI_UPST_REGION='ap-mumbai-1' +``` + +UPST authentication takes precedence over `OCI_CONFIG_PROFILE`. The server generates an +RSA signing key, obtains a UPST on the first OCI command, and automatically renews it +shortly before expiration. Credentials are stored in a private temporary directory and +removed when the server exits. To select a persistent private location, set +`OCI_UPST_CREDENTIALS_DIR`; individual paths can be overridden with +`OCI_UPST_PRIVATE_KEY_FILE`, `OCI_UPST_TOKEN_FILE`, and `OCI_UPST_CONFIG_FILE`. + +The identity represented by the UPST must have OCI IAM policies authorizing requested +operations. Do not expose the client secret, token, private key, or generated config. + ## Third-Party APIs Developers choosing to distribute a binary implementation of this project are responsible for obtaining and providing all required licenses and copyright notices for the third-party code used in order to ensure compliance with their respective open source licenses. diff --git a/src/oci-api-mcp-server/oracle/oci_api_mcp_server/__init__.py b/src/oci-api-mcp-server/oracle/oci_api_mcp_server/__init__.py index ad464632..e59c4317 100644 --- a/src/oci-api-mcp-server/oracle/oci_api_mcp_server/__init__.py +++ b/src/oci-api-mcp-server/oracle/oci_api_mcp_server/__init__.py @@ -5,4 +5,4 @@ """ __project__ = "oracle.oci-api-mcp-server" -__version__ = "2.0.0" +__version__ = "2.1.0" diff --git a/src/oci-api-mcp-server/oracle/oci_api_mcp_server/auth.py b/src/oci-api-mcp-server/oracle/oci_api_mcp_server/auth.py new file mode 100644 index 00000000..972fe53e --- /dev/null +++ b/src/oci-api-mcp-server/oracle/oci_api_mcp_server/auth.py @@ -0,0 +1,310 @@ +""" +Copyright (c) 2026, Oracle and/or its affiliates. +Licensed under the Universal Permissive License v1.0 as shown at +https://oss.oracle.com/licenses/upl. +""" + +"""JWT-to-UPST authentication support for OCI CLI invocations.""" + +import atexit +import base64 +import configparser +import io +import json +import os +import shutil +import tempfile +import threading +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Mapping +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode, urlsplit +from urllib.request import Request, urlopen + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + +_REQUIRED_ENVIRONMENT_VARIABLES = ( + "OCI_UPST_DOMAIN_URL", + "OCI_UPST_CLIENT_ID", + "OCI_UPST_CLIENT_SECRET", + "OCI_UPST_REGION", +) +_PROFILE_NAME = "UPST" +_EXPIRY_SKEW_SECONDS = 60 +_temporary_directory: Path | None = None +_upst_session: "UpstSession | None" = None +_upst_session_lock = threading.Lock() + + +class UpstAuthenticationError(RuntimeError): + """Raised when OCI JWT-to-UPST authentication cannot be completed.""" + + +@dataclass(frozen=True) +class UpstSession: + """The generated CLI configuration and optional UPST expiration time.""" + + config_file: str + profile_name: str + expires_at: int | None + + def is_usable(self, now: float) -> bool: + """Return whether the token is not within the configured renewal window.""" + return self.expires_at is None or now < self.expires_at - _EXPIRY_SKEW_SECONDS + + +def is_upst_auth_configured(environment: Mapping[str, str] | None = None) -> bool: + """Return whether any UPST-specific environment variable was supplied.""" + environment = environment or os.environ + return any(environment.get(name) for name in _REQUIRED_ENVIRONMENT_VARIABLES) + + +def get_upst_cli_configuration( + environment: Mapping[str, str] | None = None, +) -> tuple[str, str]: + """Return a UPST-backed OCI CLI configuration, renewing near token expiration.""" + global _upst_session + environment = environment or os.environ + _validate_environment(environment) + + with _upst_session_lock: + if _upst_session is not None and _upst_session.is_usable(time.time()): + return _upst_session.config_file, _upst_session.profile_name + + directory = _get_credential_directory(environment) + private_key_file = _path_from_environment( + environment, "OCI_UPST_PRIVATE_KEY_FILE", directory / "private_key.pem" + ) + token_file = _path_from_environment( + environment, "OCI_UPST_TOKEN_FILE", directory / "token" + ) + config_file = _path_from_environment( + environment, "OCI_UPST_CONFIG_FILE", directory / "config" + ) + + private_key = _load_or_create_private_key(private_key_file) + public_key = base64.b64encode( + private_key.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + ).decode("ascii") + service_bearer_token = _get_service_bearer_token(environment) + upst_token = _exchange_for_upst(environment, service_bearer_token, public_key) + expires_at = _get_jwt_expiration(upst_token) + if expires_at is not None and expires_at <= time.time(): + raise UpstAuthenticationError("UPST exchange returned an expired token.") + + _write_private_file(token_file, upst_token) + _write_cli_config( + config_file, private_key_file, token_file, environment["OCI_UPST_REGION"] + ) + _upst_session = UpstSession(str(config_file), _PROFILE_NAME, expires_at) + return _upst_session.config_file, _upst_session.profile_name + + +def _validate_environment(environment: Mapping[str, str]) -> None: + missing = [ + name for name in _REQUIRED_ENVIRONMENT_VARIABLES if not environment.get(name) + ] + if missing: + raise UpstAuthenticationError( + "UPST authentication requires environment variables: " + ", ".join(missing) + ) + parsed_url = urlsplit(environment["OCI_UPST_DOMAIN_URL"]) + if ( + parsed_url.scheme != "https" + or not parsed_url.netloc + or parsed_url.query + or parsed_url.fragment + ): + raise UpstAuthenticationError( + "OCI_UPST_DOMAIN_URL must be an HTTPS identity domain URL." + ) + + +def _get_credential_directory(environment: Mapping[str, str]) -> Path: + global _temporary_directory + if _temporary_directory is None: + configured_directory = environment.get("OCI_UPST_CREDENTIALS_DIR") + if configured_directory: + _temporary_directory = Path(os.path.expanduser(configured_directory)) + _temporary_directory.mkdir(mode=0o700, parents=True, exist_ok=True) + else: + _temporary_directory = Path(tempfile.mkdtemp(prefix="oci-upst-")) + atexit.register(shutil.rmtree, _temporary_directory, ignore_errors=True) + _require_private_directory(_temporary_directory) + return _temporary_directory + + +def _path_from_environment( + environment: Mapping[str, str], name: str, default: Path +) -> Path: + return Path(os.path.expanduser(environment.get(name, str(default)))) + + +def _require_private_directory(directory: Path) -> None: + if not directory.is_dir(): + raise UpstAuthenticationError( + f"UPST credentials directory is not a directory: {directory}" + ) + if directory.stat().st_mode & 0o077: + raise UpstAuthenticationError( + f"UPST credentials directory must not be group or world accessible: {directory}" + ) + + +def _load_or_create_private_key(private_key_file: Path) -> rsa.RSAPrivateKey: + if private_key_file.exists(): + if private_key_file.stat().st_mode & 0o077: + raise UpstAuthenticationError( + "UPST private key must not be group or world accessible." + ) + try: + private_key = serialization.load_pem_private_key( + private_key_file.read_bytes(), password=None + ) + except (TypeError, ValueError) as error: + raise UpstAuthenticationError( + "Failed to load the UPST RSA private key." + ) from error + if ( + not isinstance(private_key, rsa.RSAPrivateKey) + or private_key.key_size < 2048 + ): + raise UpstAuthenticationError( + "UPST private key must be an RSA key of at least 2048 bits." + ) + return private_key + + private_key_file.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + _write_private_file( + private_key_file, + private_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ).decode("ascii"), + ) + return private_key + + +def _get_service_bearer_token(environment: Mapping[str, str]) -> str: + response = _post_token_request( + environment, + {"grant_type": "client_credentials", "scope": "urn:opc:idm:__myscopes__"}, + ) + token = response.get("access_token") + if ( + not isinstance(token, str) + or not token + or token.count(".") != 2 + or any(token.isspace() for token in token) + ): + raise UpstAuthenticationError( + "Service-bearer response did not contain a valid JWT access_token." + ) + return token + + +def _exchange_for_upst( + environment: Mapping[str, str], service_bearer_token: str, public_key: str +) -> str: + response = _post_token_request( + environment, + { + "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", + "requested_token_type": "urn:oci:token-type:oci-upst", + "public_key": public_key, + "subject_token": service_bearer_token, + "subject_token_type": "jwt", + }, + ) + token = response.get("token") + if not isinstance(token, str) or not token: + raise UpstAuthenticationError("UPST exchange response did not contain token.") + return token + + +def _post_token_request( + environment: Mapping[str, str], payload: Mapping[str, str] +) -> dict: + endpoint = environment["OCI_UPST_DOMAIN_URL"].rstrip("/") + "/oauth2/v1/token" + credentials = ( + f"{environment['OCI_UPST_CLIENT_ID']}:{environment['OCI_UPST_CLIENT_SECRET']}" + ) + authorization = base64.b64encode(credentials.encode("utf-8")).decode("ascii") + request = Request( + endpoint, + data=urlencode(payload).encode("utf-8"), + headers={ + "Authorization": f"Basic {authorization}", + "Content-Type": "application/x-www-form-urlencoded", + }, + method="POST", + ) + try: + with urlopen(request, timeout=30) as response: + parsed_response = json.loads(response.read().decode("utf-8")) + except HTTPError as error: + raise UpstAuthenticationError( + f"OCI UPST token request failed with HTTP {error.code}." + ) from error + except (OSError, URLError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise UpstAuthenticationError("OCI UPST token request failed.") from error + if not isinstance(parsed_response, dict): + raise UpstAuthenticationError("OCI UPST token response was not a JSON object.") + return parsed_response + + +def _get_jwt_expiration(token: str) -> int | None: + parts = token.split(".") + if len(parts) != 3: + return None + try: + payload = json.loads( + base64.urlsafe_b64decode(parts[1] + "=" * (-len(parts[1]) % 4)) + ) + except (UnicodeDecodeError, ValueError, json.JSONDecodeError): + return None + expiration = payload.get("exp") if isinstance(payload, dict) else None + return int(expiration) if isinstance(expiration, int | float) else None + + +def _write_private_file(path: Path, content: str) -> None: + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + file_descriptor, temporary_path = tempfile.mkstemp( + prefix=f".{path.name}.", dir=path.parent + ) + try: + os.chmod(temporary_path, 0o600) + with os.fdopen(file_descriptor, "w", encoding="utf-8") as file: + file.write(content) + file.flush() + os.fsync(file.fileno()) + os.replace(temporary_path, path) + except OSError as error: + raise UpstAuthenticationError( + f"Failed to write UPST credential file: {path}" + ) from error + finally: + if os.path.exists(temporary_path): + os.unlink(temporary_path) + + +def _write_cli_config( + config_file: Path, private_key_file: Path, token_file: Path, region: str +) -> None: + config = configparser.ConfigParser() + config[_PROFILE_NAME] = { + "region": region, + "key_file": str(private_key_file), + "security_token_file": str(token_file), + } + output = io.StringIO() + config.write(output) + _write_private_file(config_file, output.getvalue()) diff --git a/src/oci-api-mcp-server/oracle/oci_api_mcp_server/server.py b/src/oci-api-mcp-server/oracle/oci_api_mcp_server/server.py index 60a3a1e2..699a5bd4 100644 --- a/src/oci-api-mcp-server/oracle/oci_api_mcp_server/server.py +++ b/src/oci-api-mcp-server/oracle/oci_api_mcp_server/server.py @@ -6,14 +6,20 @@ import json import os +import shlex import subprocess from logging import Logger -import shlex from typing import Annotated import oci from fastmcp import FastMCP + from oracle.oci_api_mcp_server import __project__, __version__ +from oracle.oci_api_mcp_server.auth import ( + UpstAuthenticationError, + get_upst_cli_configuration, + is_upst_auth_configured, +) from oracle.oci_api_mcp_server.denylist import Denylist from oracle.oci_api_mcp_server.utils import initAuditLogger @@ -163,8 +169,19 @@ def run_oci_command( return {"error": error_message} try: + cli_auth_args = ["--profile", profile, "--auth", "security_token"] + if is_upst_auth_configured(): + config_file, profile = get_upst_cli_configuration() + cli_auth_args = [ + "--config-file", + config_file, + "--profile", + profile, + "--auth", + "security_token", + ] result = subprocess.run( - ["oci", "--profile", profile, "--auth", "security_token"] + shlex.split(command), + ["oci"] + cli_auth_args + shlex.split(command), env=env_copy, capture_output=True, text=True, @@ -187,6 +204,9 @@ def run_oci_command( pass return response + except UpstAuthenticationError as error: + logger.error("UPST authentication failed: %s", error) + return {"command": command, "output": "", "error": str(error), "returncode": 1} except subprocess.CalledProcessError as e: return { "command": command, diff --git a/src/oci-api-mcp-server/oracle/oci_api_mcp_server/tests/test_oci_api_tools.py b/src/oci-api-mcp-server/oracle/oci_api_mcp_server/tests/test_oci_api_tools.py index cad2b654..e742e6e2 100644 --- a/src/oci-api-mcp-server/oracle/oci_api_mcp_server/tests/test_oci_api_tools.py +++ b/src/oci-api-mcp-server/oracle/oci_api_mcp_server/tests/test_oci_api_tools.py @@ -8,9 +8,12 @@ import json import subprocess from unittest.mock import ANY, MagicMock, patch +from urllib.error import HTTPError import pytest from fastmcp import Client + +import oracle.oci_api_mcp_server.auth as auth import oracle.oci_api_mcp_server.server as server from oracle.oci_api_mcp_server import __project__ from oracle.oci_api_mcp_server.denylist import Denylist @@ -32,11 +35,16 @@ async def test_get_oci_command_help_success(self, mock_run): async with Client(mcp) as client: result = ( - await client.call_tool("get_oci_command_help", {"command": "compute instance list"}) + await client.call_tool( + "get_oci_command_help", {"command": "compute instance list"} + ) ).structured_content["result"] assert result == "Help output" - assert mock_run.call_args.kwargs["env"]["OCI_SDK_APPEND_USER_AGENT"] == USER_AGENT + assert ( + mock_run.call_args.kwargs["env"]["OCI_SDK_APPEND_USER_AGENT"] + == USER_AGENT + ) mock_run.assert_called_once_with( ["oci", "compute", "instance", "list", "--help"], env=ANY, @@ -58,7 +66,9 @@ async def test_get_oci_command_help_preserves_quoted_arguments(self, mock_run): result = ( await client.call_tool( "get_oci_command_help", - {"command": 'compute instance list --display-name "Shared Services"'}, + { + "command": 'compute instance list --display-name "Shared Services"' + }, ) ).structured_content["result"] @@ -95,7 +105,9 @@ async def test_get_oci_command_help_failure(self, mock_run): async with Client(mcp) as client: result = ( - await client.call_tool("get_oci_command_help", {"command": "compute instance list"}) + await client.call_tool( + "get_oci_command_help", {"command": "compute instance list"} + ) ).structured_content["result"] assert "Error: Some error" in result @@ -112,7 +124,9 @@ async def test_run_oci_command_success(self, mock_run): mock_run.return_value = mock_result async with Client(mcp) as client: - result = (await client.call_tool("run_oci_command", {"command": command})).data + result = ( + await client.call_tool("run_oci_command", {"command": command}) + ).data assert result == { "command": command, @@ -133,7 +147,9 @@ async def test_run_oci_command_string_success(self, mock_run): mock_run.return_value = mock_result async with Client(mcp) as client: - result = (await client.call_tool("run_oci_command", {"command": command})).data + result = ( + await client.call_tool("run_oci_command", {"command": command}) + ).data assert result == { "command": command, @@ -154,7 +170,9 @@ async def test_run_oci_command_preserves_quoted_arguments(self, mock_run): mock_run.return_value = mock_result async with Client(mcp) as client: - result = (await client.call_tool("run_oci_command", {"command": command})).data + result = ( + await client.call_tool("run_oci_command", {"command": command}) + ).data assert result == { "command": command, @@ -182,6 +200,60 @@ async def test_run_oci_command_preserves_quoted_arguments(self, mock_run): shell=False, ) + @pytest.mark.asyncio + @patch("oracle.oci_api_mcp_server.server.get_upst_cli_configuration") + @patch( + "oracle.oci_api_mcp_server.server.is_upst_auth_configured", return_value=True + ) + @patch("oracle.oci_api_mcp_server.server.subprocess.run") + async def test_run_oci_command_uses_upst_profile( + self, mock_run, mock_is_upst_auth_configured, mock_get_upst_cli_configuration + ): + mock_get_upst_cli_configuration.return_value = ("/private/upst/config", "UPST") + mock_run.return_value = MagicMock(stdout="{}", stderr="", returncode=0) + + async with Client(mcp) as client: + result = ( + await client.call_tool( + "run_oci_command", {"command": "iam region list"} + ) + ).data + + assert result["output"] == {} + mock_is_upst_auth_configured.assert_called_once_with() + mock_get_upst_cli_configuration.assert_called_once_with() + assert mock_run.call_args.args[0][:7] == [ + "oci", + "--config-file", + "/private/upst/config", + "--profile", + "UPST", + "--auth", + "security_token", + ] + + @pytest.mark.asyncio + @patch("oracle.oci_api_mcp_server.server.get_upst_cli_configuration") + @patch( + "oracle.oci_api_mcp_server.server.is_upst_auth_configured", return_value=True + ) + async def test_run_oci_command_returns_upst_error( + self, mock_is_upst_auth_configured, mock_get_upst_cli_configuration + ): + mock_get_upst_cli_configuration.side_effect = auth.UpstAuthenticationError( + "token exchange failed" + ) + + async with Client(mcp) as client: + result = ( + await client.call_tool( + "run_oci_command", {"command": "iam region list"} + ) + ).data + + assert result["error"] == "token exchange failed" + assert result["returncode"] == 1 + @pytest.mark.asyncio @patch("oracle.oci_api_mcp_server.server.subprocess.run") async def test_run_oci_command_failure(self, mock_run): @@ -200,7 +272,9 @@ async def test_run_oci_command_failure(self, mock_run): ) async with Client(mcp) as client: - result = (await client.call_tool("run_oci_command", {"command": command})).data + result = ( + await client.call_tool("run_oci_command", {"command": command}) + ).data assert result == { "command": command, @@ -221,7 +295,10 @@ async def test_get_oci_commands_success(self, mock_run): result = (await client.read_resource("resource://oci-api-commands"))[0].text assert result == "OCI commands output" - assert mock_run.call_args.kwargs["env"]["OCI_SDK_APPEND_USER_AGENT"] == USER_AGENT + assert ( + mock_run.call_args.kwargs["env"]["OCI_SDK_APPEND_USER_AGENT"] + == USER_AGENT + ) mock_run.assert_called_once_with( ["oci", "--help"], env=ANY, @@ -260,7 +337,9 @@ async def test_run_oci_command_denied(self, mock_json_loads, mock_run): async with Client(mcp) as client: result = ( - await client.call_tool("run_oci_command", {"command": "compute instance terminate"}) + await client.call_tool( + "run_oci_command", {"command": "compute instance terminate"} + ) ).data assert "error" in result @@ -269,14 +348,31 @@ async def test_run_oci_command_denied(self, mock_json_loads, mock_run): @pytest.mark.parametrize( ("command", "normalized"), [ - ("compute instance terminate --instance-id ocid1.instance.oc1..example", "compute instance terminate"), - ("--debug compute instance terminate --instance-id ocid1.instance.oc1..example", "compute instance terminate"), - ("--raw-output compute instance terminate --instance-id ocid1.instance.oc1..example", "compute instance terminate"), - ("--no-retry compute instance terminate --instance-id ocid1.instance.oc1..example", "compute instance terminate"), - ("--config-file /tmp/config compute instance terminate --instance-id ocid1.instance.oc1..example", "compute instance terminate"), + ( + "compute instance terminate --instance-id ocid1.instance.oc1..example", + "compute instance terminate", + ), + ( + "--debug compute instance terminate --instance-id ocid1.instance.oc1..example", + "compute instance terminate", + ), + ( + "--raw-output compute instance terminate --instance-id ocid1.instance.oc1..example", + "compute instance terminate", + ), + ( + "--no-retry compute instance terminate --instance-id ocid1.instance.oc1..example", + "compute instance terminate", + ), + ( + "--config-file /tmp/config compute instance terminate --instance-id ocid1.instance.oc1..example", + "compute instance terminate", + ), ], ) - def test_denylist_preserves_command_words_after_global_options(self, command, normalized): + def test_denylist_preserves_command_words_after_global_options( + self, command, normalized + ): denylist = Denylist(MagicMock()) assert denylist.remove_params_from_command(command) == normalized @@ -289,7 +385,9 @@ def test_denylist_preserves_command_words_after_global_options(self, command, no "--no-retry compute instance terminate --instance-id ocid1.instance.oc1..example", ], ) - def test_denylist_blocks_destructive_commands_after_valueless_global_options(self, command): + def test_denylist_blocks_destructive_commands_after_valueless_global_options( + self, command + ): denylist = Denylist(MagicMock()) denylist.denylist = ["compute instance terminate"] @@ -307,7 +405,10 @@ def test_main_without_host_and_port(self, mock_getenv, mock_mcp_run): @patch("os.getenv") def test_main_with_host_and_port(self, mock_getenv): - mock_getenv.side_effect = lambda x: {"ORACLE_MCP_HOST": "1.2.3.4", "ORACLE_MCP_PORT": "8888"}.get(x) + mock_getenv.side_effect = lambda x: { + "ORACLE_MCP_HOST": "1.2.3.4", + "ORACLE_MCP_PORT": "8888", + }.get(x) with pytest.raises(RuntimeError, match="stdio transport only"): server.main() @@ -325,3 +426,130 @@ def test_main_with_only_port(self, mock_getenv): with pytest.raises(RuntimeError, match="stdio transport only"): server.main() + + +class TestUpstAuthentication: + @pytest.fixture(autouse=True) + def reset_upst_session(self): + auth._temporary_directory = None + auth._upst_session = None + yield + auth._temporary_directory = None + auth._upst_session = None + + @staticmethod + def environment(tmp_path): + return { + "OCI_UPST_DOMAIN_URL": "https://identity.example.com", + "OCI_UPST_CLIENT_ID": "client", + "OCI_UPST_CLIENT_SECRET": "secret", + "OCI_UPST_REGION": "ap-mumbai-1", + "OCI_UPST_CREDENTIALS_DIR": str(tmp_path / "credentials"), + } + + def test_configuration_requires_all_required_environment_variables(self): + with pytest.raises(auth.UpstAuthenticationError, match="OCI_UPST_CLIENT_ID"): + auth.get_upst_cli_configuration( + {"OCI_UPST_DOMAIN_URL": "https://identity.example.com"} + ) + + @pytest.mark.parametrize( + "domain_url", + ["http://identity.example.com", "https://identity.example.com?x=1"], + ) + def test_configuration_requires_clean_https_domain_url(self, domain_url, tmp_path): + environment = self.environment(tmp_path) + environment["OCI_UPST_DOMAIN_URL"] = domain_url + with pytest.raises(auth.UpstAuthenticationError, match="HTTPS identity domain"): + auth.get_upst_cli_configuration(environment) + + @patch("oracle.oci_api_mcp_server.auth._post_token_request") + def test_creates_private_files_and_reuses_upst_during_session( + self, mock_post, tmp_path + ): + environment = self.environment(tmp_path) + mock_post.side_effect = [ + {"access_token": "header.payload.signature"}, + {"token": "upst-token"}, + ] + + first_configuration = auth.get_upst_cli_configuration(environment) + second_configuration = auth.get_upst_cli_configuration(environment) + + credentials_directory = tmp_path / "credentials" + assert first_configuration == second_configuration + assert mock_post.call_count == 2 + assert (credentials_directory / "token").read_text() == "upst-token" + assert (credentials_directory / "private_key.pem").stat().st_mode & 0o077 == 0 + assert (credentials_directory / "config").stat().st_mode & 0o077 == 0 + + @patch("oracle.oci_api_mcp_server.auth._post_token_request") + @patch("oracle.oci_api_mcp_server.auth.time.time", return_value=1000) + def test_renews_expiring_upst(self, mock_time, mock_post, tmp_path): + environment = self.environment(tmp_path) + mock_post.side_effect = [ + {"access_token": "header.payload.signature"}, + {"token": self._jwt(1050)}, + {"access_token": "header.payload.signature"}, + {"token": self._jwt(2000)}, + ] + + auth.get_upst_cli_configuration(environment) + auth.get_upst_cli_configuration(environment) + + assert mock_post.call_count == 4 + + @staticmethod + def _jwt(expiration): + payload = json.dumps({"exp": expiration}).encode() + return ( + "header." + + __import__("base64").urlsafe_b64encode(payload).decode().rstrip("=") + + ".signature" + ) + + def test_private_key_rejects_unsafe_permissions_and_invalid_content(self, tmp_path): + private_key_file = tmp_path / "key.pem" + private_key_file.write_text("not a key") + private_key_file.chmod(0o600) + with pytest.raises(auth.UpstAuthenticationError, match="Failed to load"): + auth._load_or_create_private_key(private_key_file) + + private_key_file.chmod(0o644) + with pytest.raises( + auth.UpstAuthenticationError, match="must not be group or world" + ): + auth._load_or_create_private_key(private_key_file) + + def test_expiration_parser_handles_jwt_and_opaque_tokens(self): + assert auth._get_jwt_expiration(self._jwt(1234)) == 1234 + assert auth._get_jwt_expiration("opaque-token") is None + assert auth._get_jwt_expiration("a.bad.c") is None + + @patch("oracle.oci_api_mcp_server.auth.urlopen") + def test_post_token_request_uses_basic_auth_and_form_encoding( + self, mock_urlopen, tmp_path + ): + response = MagicMock() + response.read.return_value = b'{"token": "upst"}' + mock_urlopen.return_value.__enter__.return_value = response + + result = auth._post_token_request( + self.environment(tmp_path), {"grant_type": "client_credentials"} + ) + + assert result == {"token": "upst"} + request = mock_urlopen.call_args.args[0] + assert request.full_url == "https://identity.example.com/oauth2/v1/token" + assert request.data == b"grant_type=client_credentials" + assert request.headers["Authorization"].startswith("Basic ") + + @patch("oracle.oci_api_mcp_server.auth.urlopen") + def test_post_token_request_hides_http_error_body(self, mock_urlopen, tmp_path): + mock_urlopen.side_effect = HTTPError( + "https://identity.example.com", 401, "Unauthorized", {}, None + ) + with pytest.raises(auth.UpstAuthenticationError, match="HTTP 401"): + auth._post_token_request( + self.environment(tmp_path), {"grant_type": "client_credentials"} + ) diff --git a/src/oci-api-mcp-server/pyproject.toml b/src/oci-api-mcp-server/pyproject.toml index 54990670..65872966 100644 --- a/src/oci-api-mcp-server/pyproject.toml +++ b/src/oci-api-mcp-server/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "oracle.oci-api-mcp-server" -version = "2.0.0" +version = "2.1.0" description = "OCI CLI MCP server" readme = "README.md" requires-python = ">=3.13" @@ -10,6 +10,7 @@ authors = [ {name = "Oracle MCP", email = "237432095+oracle-mcp@users.noreply.github.com"}, ] dependencies = [ + "cryptography==46.0.7", "fastmcp==3.4.2", "oci-cli==3.87.0" ] @@ -34,6 +35,9 @@ build-backend = "hatchling.build" packages = ["oracle"] exclude = ["/oracle/**/tests/**"] +[tool.isort] +profile = "black" + [dependency-groups] dev = [ "pytest>=9.0.3", diff --git a/src/oci-api-mcp-server/uv.lock b/src/oci-api-mcp-server/uv.lock index a1ec40e0..482957de 100644 --- a/src/oci-api-mcp-server/uv.lock +++ b/src/oci-api-mcp-server/uv.lock @@ -809,9 +809,10 @@ wheels = [ [[package]] name = "oracle-oci-api-mcp-server" -version = "2.0.0" +version = "2.1.0" source = { editable = "." } dependencies = [ + { name = "cryptography" }, { name = "fastmcp" }, { name = "oci-cli" }, ] @@ -825,6 +826,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "cryptography", specifier = "==46.0.7" }, { name = "fastmcp", specifier = "==3.4.2" }, { name = "oci-cli", specifier = "==3.87.0" }, ]