mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-09-07 08:26:06 +00:00
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""Return the effective application configuration."""
|
|
|
|
from typing import Any
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
|
|
|
from ..base_step import BaseStep
|
|
from ...components import R
|
|
|
|
_SECRET_KEYS = {
|
|
"api_key",
|
|
"app_secret",
|
|
"authorization",
|
|
"cookie",
|
|
"password",
|
|
"private_key",
|
|
"proxy_authorization",
|
|
"secret",
|
|
"set_cookie",
|
|
"token",
|
|
}
|
|
_SECRET_SUFFIXES = ("_access_key", "_api_key", "_password", "_private_key", "_secret", "_token")
|
|
|
|
|
|
def _is_secret_key(key: str) -> bool:
|
|
normalized = key.lower().replace("-", "_")
|
|
return normalized in _SECRET_KEYS or normalized.endswith(_SECRET_SUFFIXES)
|
|
|
|
|
|
def _redact_url(value: str) -> str:
|
|
"""Redact credentials embedded in an HTTP-style URL."""
|
|
try:
|
|
parsed = urlsplit(value)
|
|
if not parsed.scheme or not parsed.netloc:
|
|
return value
|
|
|
|
netloc = parsed.netloc
|
|
if parsed.password is not None:
|
|
host = parsed.hostname or ""
|
|
if ":" in host:
|
|
host = f"[{host}]"
|
|
if parsed.port is not None:
|
|
host = f"{host}:{parsed.port}"
|
|
netloc = f"{parsed.username or ''}:***@{host}"
|
|
|
|
query = urlencode(
|
|
[
|
|
(key, "***" if _is_secret_key(key) else item)
|
|
for key, item in parse_qsl(parsed.query, keep_blank_values=True)
|
|
],
|
|
)
|
|
return urlunsplit((parsed.scheme, netloc, parsed.path, query, parsed.fragment))
|
|
except ValueError:
|
|
return value
|
|
|
|
|
|
def _redact(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
redacted = {}
|
|
for key, item in value.items():
|
|
key_string = str(key)
|
|
if _is_secret_key(key_string) or (
|
|
key_string.lower() == "credential" and not isinstance(item, (dict, list))
|
|
):
|
|
redacted[key] = "***"
|
|
else:
|
|
redacted[key] = _redact(item)
|
|
return redacted
|
|
if isinstance(value, list):
|
|
return [_redact(item) for item in value]
|
|
if isinstance(value, str):
|
|
return _redact_url(value)
|
|
return value
|
|
|
|
|
|
@R.register("app_config_step")
|
|
class AppConfigStep(BaseStep):
|
|
"""Return the effective config without startup environment or secrets."""
|
|
|
|
async def execute(self):
|
|
assert self.context is not None
|
|
assert self.app_context is not None
|
|
|
|
config = self.app_context.app_config.model_dump(mode="json", exclude={"environment"})
|
|
self.context.response.answer = _redact(config)
|
|
return self.context.response
|