| @@ -13,6 +13,7 @@ from .observability import ObservabilityConfig | |||
| from .packaging import PackagingInfo | |||
| from .remote_settings_sources import RemoteSettingsSource, RemoteSettingsSourceConfig, RemoteSettingsSourceName | |||
| from .remote_settings_sources.apollo import ApolloSettingsSource | |||
| from .remote_settings_sources.nacos import NacosSettingsSource | |||
| logger = logging.getLogger(__name__) | |||
| @@ -34,6 +35,8 @@ class RemoteSettingsSourceFactory(PydanticBaseSettingsSource): | |||
| match remote_source_name: | |||
| case RemoteSettingsSourceName.APOLLO: | |||
| remote_source = ApolloSettingsSource(current_state) | |||
| case RemoteSettingsSourceName.NACOS: | |||
| remote_source = NacosSettingsSource(current_state) | |||
| case _: | |||
| logger.warning(f"Unsupported remote source: {remote_source_name}") | |||
| return {} | |||
| @@ -3,3 +3,4 @@ from enum import StrEnum | |||
| class RemoteSettingsSourceName(StrEnum): | |||
| APOLLO = "apollo" | |||
| NACOS = "nacos" | |||
| @@ -0,0 +1,52 @@ | |||
| import logging | |||
| import os | |||
| from collections.abc import Mapping | |||
| from typing import Any | |||
| from pydantic.fields import FieldInfo | |||
| from .http_request import NacosHttpClient | |||
| logger = logging.getLogger(__name__) | |||
| from configs.remote_settings_sources.base import RemoteSettingsSource | |||
| from .utils import _parse_config | |||
| class NacosSettingsSource(RemoteSettingsSource): | |||
| def __init__(self, configs: Mapping[str, Any]): | |||
| self.configs = configs | |||
| self.remote_configs: dict[str, Any] = {} | |||
| self.async_init() | |||
| def async_init(self): | |||
| data_id = os.getenv("DIFY_ENV_NACOS_DATA_ID", "dify-api-env.properties") | |||
| group = os.getenv("DIFY_ENV_NACOS_GROUP", "nacos-dify") | |||
| tenant = os.getenv("DIFY_ENV_NACOS_NAMESPACE", "") | |||
| params = {"dataId": data_id, "group": group, "tenant": tenant} | |||
| try: | |||
| content = NacosHttpClient().http_request("/nacos/v1/cs/configs", method="GET", headers={}, params=params) | |||
| self.remote_configs = self._parse_config(content) | |||
| except Exception as e: | |||
| logger.exception("[get-access-token] exception occurred") | |||
| raise | |||
| def _parse_config(self, content: str) -> dict: | |||
| if not content: | |||
| return {} | |||
| try: | |||
| return _parse_config(self, content) | |||
| except Exception as e: | |||
| raise RuntimeError(f"Failed to parse config: {e}") | |||
| def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]: | |||
| if not isinstance(self.remote_configs, dict): | |||
| raise ValueError(f"remote configs is not dict, but {type(self.remote_configs)}") | |||
| field_value = self.remote_configs.get(field_name) | |||
| if field_value is None: | |||
| return None, field_name, False | |||
| return field_value, field_name, False | |||
| @@ -0,0 +1,83 @@ | |||
| import base64 | |||
| import hashlib | |||
| import hmac | |||
| import logging | |||
| import os | |||
| import time | |||
| import requests | |||
| logger = logging.getLogger(__name__) | |||
| class NacosHttpClient: | |||
| def __init__(self): | |||
| self.username = os.getenv("DIFY_ENV_NACOS_USERNAME") | |||
| self.password = os.getenv("DIFY_ENV_NACOS_PASSWORD") | |||
| self.ak = os.getenv("DIFY_ENV_NACOS_ACCESS_KEY") | |||
| self.sk = os.getenv("DIFY_ENV_NACOS_SECRET_KEY") | |||
| self.server = os.getenv("DIFY_ENV_NACOS_SERVER_ADDR", "localhost:8848") | |||
| self.token = None | |||
| self.token_ttl = 18000 | |||
| self.token_expire_time: float = 0 | |||
| def http_request(self, url, method="GET", headers=None, params=None): | |||
| try: | |||
| self._inject_auth_info(headers, params) | |||
| response = requests.request(method, url="http://" + self.server + url, headers=headers, params=params) | |||
| response.raise_for_status() | |||
| return response.text | |||
| except requests.exceptions.RequestException as e: | |||
| return f"Request to Nacos failed: {e}" | |||
| def _inject_auth_info(self, headers, params, module="config"): | |||
| headers.update({"User-Agent": "Nacos-Http-Client-In-Dify:v0.0.1"}) | |||
| if module == "login": | |||
| return | |||
| ts = str(int(time.time() * 1000)) | |||
| if self.ak and self.sk: | |||
| sign_str = self.get_sign_str(params["group"], params["tenant"], ts) | |||
| headers["Spas-AccessKey"] = self.ak | |||
| headers["Spas-Signature"] = self.__do_sign(sign_str, self.sk) | |||
| headers["timeStamp"] = ts | |||
| if self.username and self.password: | |||
| self.get_access_token(force_refresh=False) | |||
| params["accessToken"] = self.token | |||
| def __do_sign(self, sign_str, sk): | |||
| return ( | |||
| base64.encodebytes(hmac.new(sk.encode(), sign_str.encode(), digestmod=hashlib.sha1).digest()) | |||
| .decode() | |||
| .strip() | |||
| ) | |||
| def get_sign_str(self, group, tenant, ts): | |||
| sign_str = "" | |||
| if tenant: | |||
| sign_str = tenant + "+" | |||
| if group: | |||
| sign_str = sign_str + group + "+" | |||
| if sign_str: | |||
| sign_str += ts | |||
| return sign_str | |||
| def get_access_token(self, force_refresh=False): | |||
| current_time = time.time() | |||
| if self.token and not force_refresh and self.token_expire_time > current_time: | |||
| return self.token | |||
| params = {"username": self.username, "password": self.password} | |||
| url = "http://" + self.server + "/nacos/v1/auth/login" | |||
| try: | |||
| resp = requests.request("POST", url, headers=None, params=params) | |||
| resp.raise_for_status() | |||
| response_data = resp.json() | |||
| self.token = response_data.get("accessToken") | |||
| self.token_ttl = response_data.get("tokenTtl", 18000) | |||
| self.token_expire_time = current_time + self.token_ttl - 10 | |||
| except Exception as e: | |||
| logger.exception("[get-access-token] exception occur") | |||
| raise | |||
| @@ -0,0 +1,31 @@ | |||
| def _parse_config(self, content: str) -> dict[str, str]: | |||
| config: dict[str, str] = {} | |||
| if not content: | |||
| return config | |||
| for line in content.splitlines(): | |||
| cleaned_line = line.strip() | |||
| if not cleaned_line or cleaned_line.startswith(("#", "!")): | |||
| continue | |||
| separator_index = -1 | |||
| for i, c in enumerate(cleaned_line): | |||
| if c in ("=", ":") and (i == 0 or cleaned_line[i - 1] != "\\"): | |||
| separator_index = i | |||
| break | |||
| if separator_index == -1: | |||
| continue | |||
| key = cleaned_line[:separator_index].strip() | |||
| raw_value = cleaned_line[separator_index + 1 :].strip() | |||
| try: | |||
| decoded_value = bytes(raw_value, "utf-8").decode("unicode_escape") | |||
| decoded_value = decoded_value.replace(r"\=", "=").replace(r"\:", ":") | |||
| except UnicodeDecodeError: | |||
| decoded_value = raw_value | |||
| config[key] = decoded_value | |||
| return config | |||