joungmin/vpd-permission-poc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refs #740: enforce carrier VPD by MCP user

Browse Source
This commit is contained in:
devmrko
2026-08-10 17:26:01 +09:00
parent 2f5fc2bfbd
commit 6394bbf078
11 changed files with 521 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
ai-web-agent-console/.env.example
View File

@@ -3,6 +3,11 @@
# MCP
HMM_MCP_BEARER_TOKEN=
HMM_MCP_BEARER_TOKEN_E1001=
HMM_MCP_BEARER_TOKEN_E1002=
HMM_MCP_BEARER_TOKEN_E1003=
HMM_MCP_BEARER_TOKEN_E1005=
HMM_MCP_BEARER_TOKEN_E1007=
AI_WEB_AGENT_CONSOLE_MCP_TIMEOUT_SECONDS=45
# OCI Generative AI SDK

4
ai-web-agent-console/README.md
View File

@@ -41,7 +41,9 @@ python3 -m venv .venv
- `config/hmm_hr_query_contracts.json`: 질의별 필수 근거와 계산·시간 규칙
실제 토큰, DB 비밀번호, Wallet, OCI private key는 Git에 넣지 않는다. 데모 사용자 JSON은
`HMM_MCP_BEARER_TOKEN` 같은 환경변수 이름만 참조한다.
`HMM_MCP_BEARER_TOKEN_E1001`처럼 사용자별 환경변수 이름만 참조한다. token 원문은 설정 JSON이나
문서에 기록하지 않는다. 사용자 preset들이 같은 환경변수를 공유하면 화면의 사용자만 바뀌고 DB
권한 문맥은 바뀌지 않으므로 허용하지 않는다.
## 검증

12
ai-web-agent-console/app.py
View File

@@ -2864,10 +2864,10 @@ def _clean_agent_tool_query(value: object, fallback: str) -> str:
def _camel_case_key(value: object) -> str:
text = str(value or "")
if "_" not in text:
text = str(value or "").strip()
if not re.search(r"[\s_-]", text):
return text[:1].lower() + text[1:]
parts = [part for part in text.casefold().split("_") if part]
parts = [part for part in re.split(r"[\s_-]+", text.casefold()) if part]
return (
parts[0] + "".join(part[:1].upper() + part[1:] for part in parts[1:])
if parts
@@ -2897,11 +2897,11 @@ def _clean_presentation_title(value: Any) -> str:
def _fallback_presentation_data_query(question: str) -> str:
text = " ".join(str(question or "").split()).strip()
text = re.sub(
r"\s*(?:그리고\s*)?(?:이걸|이를|그\s*결과를|결과를)\s*"
r"\s*(?:그리고\s*)?(?:(?:이걸|이를|그\s*결과를|결과를)\s*)?"
r"(?:HMM\s*)?(?:HTML|리포트|보고서|대시보드|차트)\s*"
r"(?:형식)?(?:으로|로)?\s*(?:만들어|생성해|작성해|보여)\s*"
r"(?:줘|주세요)?\s*[.!?]?\s*$",
"",
" 보여줘",
text,
flags=re.IGNORECASE,
)
@@ -2912,7 +2912,7 @@ def _fallback_presentation_data_query(question: str) -> str:
)
text = re.sub(
r"(?:HMM\s*)?(?:리포트|보고서|대시보드|차트)\s*"
r"(?:형식)?(?:으로|로)?\s*(?:만들어|생성해|작성해)\s*"
r"(?:형식)?(?:으로|로)?\s*(?:만들어|생성해|작성해|보여)\s*"
r"(?:줘|주세요)?",
"보여줘",
text,

10
ai-web-agent-console/config/vpd_token_presets.json
View File

@@ -5,7 +5,7 @@
{
"enabled": true,
"default": true,
"mcp_token_env": "HMM_MCP_BEARER_TOKEN",
"mcp_token_env": "HMM_MCP_BEARER_TOKEN_E1001",
"user_id": "E1001",
"name": "Kim Minseo",
"role": "HR Team Manager",
@@ -14,7 +14,7 @@
},
{
"enabled": true,
"mcp_token_env": "HMM_MCP_BEARER_TOKEN",
"mcp_token_env": "HMM_MCP_BEARER_TOKEN_E1002",
"user_id": "E1002",
"name": "Lee Jiwon",
"role": "HR Operations Specialist",
@@ -23,7 +23,7 @@
},
{
"enabled": true,
"mcp_token_env": "HMM_MCP_BEARER_TOKEN",
"mcp_token_env": "HMM_MCP_BEARER_TOKEN_E1003",
"user_id": "E1003",
"name": "Park Dohyun",
"role": "People Analytics Analyst",
@@ -32,7 +32,7 @@
},
{
"enabled": true,
"mcp_token_env": "HMM_MCP_BEARER_TOKEN",
"mcp_token_env": "HMM_MCP_BEARER_TOKEN_E1005",
"user_id": "E1005",
"name": "Han Seojun",
"role": "Recruiting Specialist",
@@ -41,7 +41,7 @@
},
{
"enabled": true,
"mcp_token_env": "HMM_MCP_BEARER_TOKEN",
"mcp_token_env": "HMM_MCP_BEARER_TOKEN_E1007",
"user_id": "E1007",
"name": "Kang Minho",
"role": "HR Coordinator",

36
ai-web-agent-console/tests/test_scenarios.py
View File

@@ -176,6 +176,32 @@ class DemoScenarioConfigTest(unittest.TestCase):
self.assertEqual(len(rows), 2)
self.assertEqual(rows[0]["CARRIER_CODE"], "C901")
def test_hmm_report_normalizes_repeated_select_ai_column_labels(self) -> None:
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
tree = ast.parse(source)
helpers = [
node
for node in tree.body
if isinstance(node, ast.FunctionDef)
and node.name in {"_camel_case_key", "_normalize_presentation_value"}
]
namespace: dict[str, Any] = {
"Any": Any,
"Mapping": Mapping,
"re": re,
}
exec(compile(ast.Module(body=helpers, type_ignores=[]), "app.py", "exec"), namespace)
first = namespace["_normalize_presentation_value"](
{"CARRIER_CODE": "C001", "LATEST_REVENUE_USD": 100}
)
repeated = namespace["_normalize_presentation_value"](
{"carrier Code": "C002", "LATEST REVENUE USD": 200}
)
self.assertEqual(first, {"carrierCode": "C001", "latestRevenueUsd": 100})
self.assertEqual(repeated, {"carrierCode": "C002", "latestRevenueUsd": 200})
def test_hmm_report_title_and_answer_follow_presentation_contract(self) -> None:
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
tree = ast.parse(source)
@@ -235,6 +261,11 @@ class DemoScenarioConfigTest(unittest.TestCase):
self.assertIn("위험 등급", query)
self.assertIn("_mcp_rows_contain_presentation_markup(mcp_result)", source)
personal_query = namespace["_fallback_presentation_data_query"](
"내 담당 선사와 최신 매출을 리포트로 보여줘"
)
self.assertEqual(personal_query, "내 담당 선사와 최신 매출을 보여줘")
def test_hmm_report_template_contains_only_dynamic_payload_slot(self) -> None:
template = (
Path(__file__).parents[1]
@@ -255,7 +286,10 @@ class DemoScenarioConfigTest(unittest.TestCase):
self.assertEqual({item["user_id"] for item in presets}, {
"E1001", "E1002", "E1003", "E1005", "E1007"
})
self.assertTrue(all(item["mcp_token_env"] == "HMM_MCP_BEARER_TOKEN" for item in presets))
self.assertEqual(
{item["mcp_token_env"] for item in presets},
{f"HMM_MCP_BEARER_TOKEN_{item['user_id']}" for item in presets},
)
self.assertTrue(all("token" not in item for item in presets))
def test_duplicate_id_is_rejected(self) -> None: