From ce5db9f1280b71cb1c969d71d3908a975febe4f9 Mon Sep 17 00:00:00 2001
From: wxg0103 <727495428@qq.com>
Date: Wed, 15 Jul 2026 14:57:55 +0800
Subject: [PATCH] fix: add validation for AWS profile names and credential
 values in AWS credential management

---
 .../aws_bedrock_model_provider/model/llm.py   | 45 +++++++++++++++----
 1 file changed, 36 insertions(+), 9 deletions(-)

diff --git a/apps/models_provider/impl/aws_bedrock_model_provider/model/llm.py b/apps/models_provider/impl/aws_bedrock_model_provider/model/llm.py
index 50ee4abfe65..8806575bfa9 100644
--- a/apps/models_provider/impl/aws_bedrock_model_provider/model/llm.py
+++ b/apps/models_provider/impl/aws_bedrock_model_provider/model/llm.py
@@ -1,3 +1,4 @@
+import configparser
 import os
 import re
 from typing import Dict, List
@@ -77,27 +78,53 @@ def new_instance(cls, model_type: str, model_name: str, model_credential: Dict[s
     def get_num_tokens_from_messages(self, messages: List[BaseMessage]) -> int:
         try:
             return super().get_num_tokens_from_messages(messages)
-        except Exception as e:
+        except Exception:
             tokenizer = TokenizerManage.get_tokenizer()
             return sum([len(tokenizer.encode(get_buffer_string([m]))) for m in messages])
 
     def get_num_tokens(self, text: str) -> int:
         try:
             return super().get_num_tokens(text)
-        except Exception as e:
+        except Exception:
             tokenizer = TokenizerManage.get_tokenizer()
             return len(tokenizer.encode(text))
 
+def _validate_aws_profile_name(profile_name: str):
+    if not profile_name:
+        raise ValueError("profile_name can not be empty")
+    if profile_name != profile_name.strip():
+        raise ValueError("profile_name can not contain leading or trailing spaces")
+    if re.search(r'[\x00-\x1f\x7f\[\]]', profile_name):
+        raise ValueError("profile_name contains invalid characters")
+
+
+def _validate_aws_credential_value(field_name: str, value: str):
+    if not value:
+        raise ValueError(f"{field_name} can not be empty")
+    if value.strip() != value:
+        raise ValueError(f"{field_name} can not contain leading or trailing spaces")
+    if re.search(r'[\x00-\x1f\x7f]', value):
+        raise ValueError(f"{field_name} contains invalid characters")
+
+
 def _update_aws_credentials(profile_name, access_key_id, secret_access_key):
+    _validate_aws_profile_name(profile_name)
+    _validate_aws_credential_value("access_key_id", access_key_id)
+    _validate_aws_credential_value("secret_access_key", secret_access_key)
+
     credentials_path = os.path.join(os.path.expanduser("~"), ".aws", "credentials")
     os.makedirs(os.path.dirname(credentials_path), exist_ok=True)
+    config = configparser.RawConfigParser(strict=False)
+    config.optionxform = str
+    if os.path.exists(credentials_path):
+        config.read(credentials_path, encoding='utf-8')
 
-    content = open(credentials_path, 'r').read() if os.path.exists(credentials_path) else ''
-    pattern = rf'\n*\[{profile_name}\]\n*(aws_access_key_id = .*)\n*(aws_secret_access_key = .*)\n*'
-    content = re.sub(pattern, '', content, flags=re.DOTALL)
+    if config.has_section(profile_name):
+        config.remove_section(profile_name)
 
-    if not re.search(rf'\[{profile_name}\]', content):
-        content += f"\n[{profile_name}]\naws_access_key_id = {access_key_id}\naws_secret_access_key = {secret_access_key}\n"
+    config.add_section(profile_name)
+    config.set(profile_name, "aws_access_key_id", access_key_id)
+    config.set(profile_name, "aws_secret_access_key", secret_access_key)
 
-    with open(credentials_path, 'w') as file:
-        file.write(content)
+    with open(credentials_path, 'w', encoding='utf-8') as file:
+        config.write(file)