Mayne0213/mas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FIX(app): improve research_agent information retrieval

Browse Source 
- Change execute_host to execute_bash (run kubectl in container)
- Return natural language instead of JSON in Information Query mode
- Add command guide for storage/memory distinction
- Improve to user-friendly response format
This commit is contained in:
Mayne0213
2025-12-24 23:22:01 +09:00
parent beee47c6ec
commit e8dc4ff450
7 changed files with 135 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -16,6 +16,7 @@ htmlcov/
dist/
build/
*.egg-info/
deploy/docker/docker-compose.yml
# Environment variables
.env

73
deploy/docker/docker-compose.yml
View File

@@ -1,73 +0,0 @@
version: '3.8'
services:
mas:
build: ../../services/backend
container_name: mas
ports:
- "8000:8000"
environment:
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
# Groq API (OpenAI-compatible)
- GROQ_API_KEY=${GROQ_API_KEY}
- GROQ_API_BASE=${GROQ_API_BASE:-https://api.groq.com/openai/v1}
# (optional) keep other providers
- OPENAI_API_KEY=${OPENAI_API_KEY}
- GOOGLE_API_KEY=${GOOGLE_API_KEY}
- DATABASE_URL=postgresql+asyncpg://mas:mas@postgres:5432/mas
- REDIS_URL=redis://redis:6379/0
depends_on:
- redis
- postgres
- ollama
volumes:
- ../../services/backend:/app
networks:
- mas-network
# Ollama (로컬 Qwen 모델)
ollama:
image: ollama/ollama:latest
container_name: mas-ollama
ports:
- "11434:11434"
volumes:
- ollama-data:/root/.ollama
networks:
- mas-network
# PostgreSQL
postgres:
image: postgres:16-alpine
container_name: mas-postgres
environment:
POSTGRES_DB: mas
POSTGRES_USER: mas
POSTGRES_PASSWORD: mas
ports:
- "5432:5432"
volumes:
- postgres-data:/var/lib/postgresql/data
networks:
- mas-network
# Redis
redis:
image: redis:7-alpine
container_name: mas-redis
ports:
- "6379:6379"
volumes:
- redis-data:/data
networks:
- mas-network
volumes:
ollama-data:
postgres-data:
redis-data:
networks:
mas-network:
driver: bridge

2
deploy/k8s/overlays/prod/kustomization.yaml
View File

@@ -14,5 +14,5 @@ commonLabels:
# 이미지 태그 설정 (ArgoCD Image Updater가 자동으로 업데이트)
images:
- name: gitea0213.kro.kr/bluemayne/mas
newTag: main-sha-83c852831c0f4ec6a4ee69f7c99d4d1277271975
newTag: main-sha-de29acaace047feffd9c49df79e7790b36aef91f

368
services/backend/PERMISSIONS.md
View File

@@ -1,368 +0,0 @@
# MAS v2.0 권한 보고서
## 📋 에이전트별 권한 요약
### ✅ FULL WRITE ACCESS (bash_tools 사용)
모든 주요 에이전트가 `execute_bash` 도구를 통해 **무제한 write 권한**을 가지고 있습니다.
| 에이전트 | 모델 | bash_tools | 주요 권한 |
|----------|------|------------|-----------|
| **Orchestrator** | Claude 4.5 | ✅ | 모든 시스템 조회/검증, 긴급 직접 실행 |
| **Planning Agent** | Claude 4.5 | ❌ | 계획 수립만 (write 불필요) |
| **Research Agent** | Groq | ✅ | K8s, DB, Git, 파일 시스템 조회 |
| **Backend Agent** | Groq | ✅ | 파일 생성, Git 커밋, DB 마이그레이션 |
| **Frontend Agent** | Groq | ✅ | 컴포넌트/스타일 파일 생성, 빌드 |
| **Infrastructure Agent** | Groq | ✅ | YAML 생성, kubectl apply, 배포 |
| **Review Agent** | Claude | ✅ | 테스트 실행, 린터 실행, 배포 확인 |
## 🔧 bash_tool 권한 범위
### execute_bash 함수 분석
```python
def execute_bash(command: str, timeout: int = 30, cwd: Optional[str] = None) -> str:
result = subprocess.run(
command,
shell=True, # ⚠️ 무제한 bash 접근
capture_output=True,
text=True,
timeout=timeout,
cwd=cwd
)
```
**특징:**
- ❌ **샌드박스 없음**: `shell=True`로 모든 bash 명령어 실행 가능
- ❌ **권한 제한 없음**: sudo, rm -rf, dd 등 위험한 명령어도 실행 가능
- ❌ **경로 제한 없음**: 모든 파일 시스템 경로 접근 가능
- ✅ **타임아웃 설정**: 기본 30초, 최대 무제한 (파라미터로 조정 가능)
- ✅ **작업 디렉토리 지정**: `cwd` 파라미터로 실행 위치 제어 가능
### 실행 가능한 작업 예시
#### ✅ 조회 (Read)
```bash
# Kubernetes
kubectl get pods -n mas
kubectl describe deployment myapp
# PostgreSQL
psql -U bluemayne -d postgres -c "SELECT * FROM users"
# Git
git log -10 --oneline
git status
# 파일 시스템
ls -la /app/repos/
cat /app/repos/project/README.md
find /app/repos -name "*.yaml"
# Prometheus
curl http://prometheus:9090/api/v1/query?query=up
```
#### ✅ 생성/수정 (Write)
```bash
# 파일 생성
cat > /app/repos/project/api/users.py << EOF
code content here
EOF
# 디렉토리 생성
mkdir -p /app/repos/project/models
# Git 작업
cd /app/repos/project && git add . && git commit -m "Add feature"
cd /app/repos/project && git push origin main
# Kubernetes 배포
kubectl apply -f /app/repos/infrastructure/deployment.yaml
kubectl delete pod failing-pod -n mas
# Database 마이그레이션
cd /app/repos/project && alembic upgrade head
# Docker 빌드
docker build -t myapp:latest /app/repos/project
# 파일 수정
sed -i 's/old/new/g' /app/repos/project/config.yaml
```
#### ⚠️ 위험한 작업 (가능하지만 주의 필요)
```bash
# 파일 삭제
rm -rf /app/repos/old-project
# 권한 변경
chmod 777 /app/repos/project
# 시스템 명령어
sudo systemctl restart service
kill -9 <pid>
# 환경 변수 조작
export SECRET_KEY=new_value
```
## 🛡️ 보안 고려사항
### 현재 상태
- **보안 수준**: ⚠️ 낮음
- **이유**: 모든 에이전트가 무제한 bash 접근 권한 보유
- **리스크**:
- LLM이 잘못된 명령어 생성 시 시스템 손상 가능
- Prompt Injection 공격에 취약
- 민감 정보 노출 가능
### 권장 보안 개선사항
#### 1. 경로 화이트리스트 (우선순위: 높음)
```python
ALLOWED_PATHS = [
"/app/repos/",
"/tmp/mas/",
"/var/log/mas/"
]
def is_safe_path(command: str) -> bool:
# 명령어에서 경로 추출 및 검증
return any(path in command for path in ALLOWED_PATHS)
```
#### 2. 명령어 블랙리스트 (우선순위: 높음)
```python
DANGEROUS_COMMANDS = [
"rm -rf /",
"dd if=/dev/zero",
":(){ :|:& };:", # Fork bomb
"chmod 777",
"sudo",
]
```
#### 3. 역할별 권한 분리 (우선순위: 중간)
```python
AGENT_PERMISSIONS = {
"orchestrator": ["read", "write"],
"planning": [], # No bash access needed
"research": ["read"],
"code_backend": ["read", "write"],
"code_frontend": ["read", "write"],
"code_infrastructure": ["read", "write", "kubectl"],
"review": ["read", "test"],
}
```
#### 4. 승인 워크플로우 (우선순위: 낮음)
```python
# 위험한 작업은 사용자 승인 필요
REQUIRES_APPROVAL = [
"kubectl delete",
"git push",
"docker build",
"rm -r",
]
```
## 📊 에이전트별 상세 권한
### 1. Orchestrator (Claude 4.5) ✅
**권한**: Full Write Access
**목적**:
- 긴급 상황 대응
- 빠른 상태 확인
- 다른 에이전트 실패 시 직접 실행
**사용 예시**:
```bash
kubectl get pods -n mas # Pod 상태 확인
git status # Git 상태 확인
cat /app/repos/project/README.md # 빠른 파일 조회
```
---
### 2. Planning Agent (Claude 4.5) ❌
**권한**: No bash access
**이유**: 계획 수립만 수행, 실행 권한 불필요
---
### 3. Research Agent (Groq) ✅
**권한**: Full Write Access (주로 Read 사용)
**목적**:
- Kubernetes 클러스터 상태 조회
- PostgreSQL 데이터베이스 탐색
- Git 레포지토리 분석
- 파일 시스템 검색
- Prometheus 메트릭 수집
**사용 예시**:
```bash
# K8s 조회
kubectl get deployments -A
kubectl describe pod myapp-123 -n mas
# DB 조회
psql -U bluemayne -d postgres -c "\dt"
psql -U bluemayne -d postgres -c "SELECT * FROM users LIMIT 10"
# Git 조회
git log -10 --oneline
git diff main..feature-branch
# 파일 검색
find /app/repos -name "*.yaml"
grep -r "API_KEY" /app/repos/project/
```
---
### 4. Backend Agent (Groq) ✅
**권한**: Full Write Access
**목적**:
- FastAPI/Node.js 코드 작성
- 데이터베이스 마이그레이션
- API 파일 생성
- Git 커밋
**사용 예시**:
```bash
# 파일 생성
cat > /app/repos/project/api/users.py << 'EOF'
from fastapi import APIRouter
router = APIRouter()
EOF
# DB 마이그레이션
cd /app/repos/project && alembic upgrade head
# Git 커밋
cd /app/repos/project && git add . && git commit -m "Add user API"
# 테스트 실행
cd /app/repos/project && pytest tests/
```
---
### 5. Frontend Agent (Groq) ✅
**권한**: Full Write Access
**목적**:
- React/Next.js 컴포넌트 작성
- CSS/Tailwind 스타일 파일 생성
- 빌드 검증
**사용 예시**:
```bash
# 컴포넌트 생성
cat > /app/repos/project/src/components/UserCard.tsx << 'EOF'
export default function UserCard() { ... }
EOF
# 스타일 생성
cat > /app/repos/project/src/styles/UserCard.module.css << 'EOF'
.card { ... }
EOF
# 빌드 테스트
cd /app/repos/project && npm run build
cd /app/repos/project && npm test
```
---
### 6. Infrastructure Agent (Groq) ✅
**권한**: Full Write Access
**목적**:
- Kubernetes YAML 생성
- kubectl apply 실행
- Docker 빌드
- ArgoCD 설정
**사용 예시**:
```bash
# YAML 생성
cat > /app/repos/infrastructure/apps/myapp/deployment.yaml << 'EOF'
apiVersion: apps/v1
kind: Deployment
...
EOF
# Kubernetes 배포
kubectl apply -f /app/repos/infrastructure/apps/myapp/
# Docker 빌드
docker build -t gitea0213.kro.kr/bluemayne/myapp:latest .
docker push gitea0213.kro.kr/bluemayne/myapp:latest
# ArgoCD 동기화
kubectl apply -f /app/repos/infrastructure/argocd/myapp.yaml
```
---
### 7. Review Agent (Claude) ✅
**권한**: Full Write Access (주로 Test 실행)
**목적**:
- 테스트 실행
- 린터 실행
- 빌드 검증
- 배포 확인
**사용 예시**:
```bash
# 테스트 실행
cd /app/repos/project && pytest tests/ -v
cd /app/repos/project && npm test
# 린터 실행
cd /app/repos/project && pylint src/
cd /app/repos/project && eslint src/
# 빌드 검증
cd /app/repos/project && docker build -t test:latest .
# 배포 확인
kubectl get pods -n mas
kubectl logs myapp-123 -n mas --tail=50
```
## 🎯 결론
### ✅ 모든 에이전트에 Write 권한 부여 완료
1. **Orchestrator**: ✅ bash_tools 추가 완료
2. **Planning**: ⚠️ 계획 수립만 수행, write 불필요
3. **Research**: ✅ 기존에 보유
4. **Backend**: ✅ 기존에 보유
5. **Frontend**: ✅ 기존에 보유
6. **Infrastructure**: ✅ 기존에 보유
7. **Review**: ✅ bash_tools 추가 완료
### 📈 권한 통계
- **bash_tools 보유**: 6/7 에이전트 (86%)
- **Write 작업 가능**: 6개 에이전트
- **Read 전용**: 0개 (Research도 write 권한 보유)
- **권한 없음**: 1개 (Planning Agent - 의도적)
### ⚠️ 주의사항
- 현재 **샌드박스 없음**, 모든 bash 명령어 실행 가능
- LLM의 올바른 프롬프트 엔지니어링이 보안의 핵심
- 프로덕션 환경에서는 위 보안 개선사항 적용 권장
---
**생성일**: 2024-12-24
**버전**: v2.0
**상태**: ✅ 모든 에이전트 write 권한 확인 완료

182
services/backend/agents/research_agent.py
View File

@@ -25,10 +25,10 @@ RESEARCH_PROMPT = """Research Agent: Analyze cluster or retrieve information.
## Two Modes
### Mode 1: Information Query (정보 조회)
User wants specific information (password, status, list, etc.)
- Execute the requested kubectl command
- Return the result directly
- No analysis needed
User wants specific information (password, status, list, storage capacity, etc.)
- Execute kubectl commands to get the information
- Provide a clear, natural language answer
- Focus on exactly what the user asked
### Mode 2: Deployment Analysis (배포 분석)
User wants deployment decision
@@ -37,21 +37,24 @@ User wants deployment decision
- Provide structured findings
## Request commands in JSON:
{"commands": [{"tool": "execute_host", "command": "kubectl get nodes", "use_sudo": true}]}
{"commands": [{"tool": "execute_bash", "command": "kubectl get nodes"}]}
Rules:
- Request 1-2 commands at a time
- Use execute_host for kubectl commands (with use_sudo: true)
- Use execute_bash for kubectl commands (kubectl is installed in the container)
- Output ONLY JSON when requesting commands
- For storage queries, use: kubectl get pvc, df -h, du -sh
- For memory queries, use: kubectl top nodes, kubectl top pods
- Be precise: storage ≠ memory
## Final report format
### For Information Query:
{
"summary": "정보 조회 완료",
"result": "actual command result",
"findings": [{"category": "조회 결과", "data": "..."}]
}
### For Information Query (IMPORTANT - Answer in natural Korean, NOT JSON):
Provide a direct answer in natural Korean language. Examples:
- "Gitea의 공유 스토리지는 10GB 할당되어 있으며, 현재 약 3.2GB를 사용 중입니다."
- "현재 클러스터에는 3개의 노드가 실행 중입니다."
DO NOT use JSON format for information queries. Just answer naturally.
### For Deployment Analysis:
{
@@ -119,9 +122,11 @@ def research_node(state: AgentState) -> AgentState:
response_text = response.content
print(f"Response: {response_text[:500]}...")
print(f"\n📝 Full Response:\n{response_text}\n") # 디버깅용 전체 응답 출력
# JSON 명령어 추출 시도
commands_executed = False
is_final_answer = False
# 방법 1: ```json ... ``` 블록에서 추출
json_match = re.search(r'```json\s*(\{.*?\})\s*```', response_text, re.DOTALL)
@@ -168,11 +173,20 @@ def research_node(state: AgentState) -> AgentState:
# 결과를 대화에 추가 (최신 것만 유지)
results_text = "\n\n".join(results)
tool_outputs.append(results_text)
# 요청 유형에 따라 다른 지시
if request_type == "information_query":
# 정보 조회: 자연어로 답변 지시
next_instruction = f"명령어 실행 결과:\n\n{results_text}\n\n**이제 위 결과를 바탕으로 사용자의 질문에 자연스러운 한국어로 답변해주세요. JSON이 아닌 일반 문장으로 작성하세요. 핵심 정보만 간결하게 전달하세요.**"
else:
# 배포 분석: 선택권 제공
next_instruction = f"명령어 실행 결과:\n\n{results_text}\n\n계속 정보가 필요하면 추가 명령어를 요청하고, 충분한 정보를 수집했으면 최종 리포트를 JSON으로 제공해주세요."
# 전체 히스토리 대신 시스템 프롬프트 + 초기 요청 + 최신 결과만 유지
conversation = [
SystemMessage(content=RESEARCH_PROMPT),
HumanMessage(content=research_request),
HumanMessage(content=f"명령어 실행 결과:\n\n{results_text}\n\n계속 정보가 필요하면 추가 명령어를 요청하고, 충분한 정보를 수집했으면 최종 리포트를 JSON으로 제공해주세요.")
HumanMessage(content=next_instruction)
]
continue # 다음 반복으로
@@ -180,59 +194,34 @@ def research_node(state: AgentState) -> AgentState:
# 최종 리포트인 경우
elif "summary" in commands_data and "findings" in commands_data:
print("\n✅ 최종 리포트 수신")
is_final_answer = True
# 요청 유형에 따라 다른 포맷
if request_type == "information_query":
# 정보 조회: 결과만 간단히 표시
# 정보 조회: result 필드가 있으면 그것을 자연어 답변으로 사용
result = commands_data.get("result", "")
findings = commands_data.get("findings", [])
summary_parts = ["✅ 조회 완료\n"]
# 조회 결과
if result:
summary_parts.append(f"**결과:**\n```\n{result}\n```")
elif findings:
# result가 있으면 그대로 사용 (자연어 답변)
final_content = result.strip()
else:
# result가 없으면 findings에서 추출
findings = commands_data.get("findings", [])
summary_parts = []
for finding in findings[:3]:
data = finding.get("data", "")
if data:
summary_parts.append(f"{data}")
final_content = "\n".join(summary_parts)
summary_parts.append(data)
final_content = "\n".join(summary_parts) if summary_parts else "정보를 찾을 수 없습니다."
# 정보 조회는 바로 종료
state["current_agent"] = "end"
else:
# 배포 분석: 상세 정보 표시
cluster_info = commands_data.get("cluster_info", {})
findings = commands_data.get("findings", [])
# 배포 분석: 간단한 상태만 표시 (Decision agent가 상세 결과 표시)
final_content = "✅ 분석 완료"
summary_parts = ["✅ 분석 완료\n"]
# 클러스터 정보
if cluster_info:
summary_parts.append("**클러스터 정보**")
if cluster_info.get("k8s_version"):
summary_parts.append(f"- Kubernetes: {cluster_info['k8s_version']}")
if cluster_info.get("nodes"):
summary_parts.append(f"- 노드: {cluster_info['nodes']}")
if cluster_info.get("existing_tools"):
tools = ", ".join(cluster_info['existing_tools'])
summary_parts.append(f"- 기존 도구: {tools}")
# 주요 발견사항
if findings:
summary_parts.append("\n**주요 발견사항**")
for finding in findings[:5]: # 최대 5개만
category = finding.get("category", "")
data = finding.get("data", "")
if category and data:
summary_parts.append(f"- {category}: {data}")
final_content = "\n".join(summary_parts)
# 배포 분석은 orchestrator로 돌아감
# 배포 분석은 orchestrator로 돌아감 (decision으로 이동)
state["current_agent"] = "orchestrator"
state["research_data"] = commands_data
@@ -245,39 +234,110 @@ def research_node(state: AgentState) -> AgentState:
except json.JSONDecodeError as e:
print(f"⚠️ JSON 파싱 실패: {e}")
# 명령어도 없고 최종 리포트도 아니면 종료
if not commands_executed:
print("\n✅ 명령어 요청 없음, 종료")
# 명령어도 없고 최종 리포트도 아니면 자연어 답변으로 간주
if not commands_executed and not is_final_answer:
print("\n✅ 자연어 답변 수신")
# 간단한 요약만 표시
content = "✅ 분석 완료\n\n기본 정보가 수집되었습니다."
# 요청 유형에 따라 다른 출력
if request_type == "information_query":
# 정보 조회: Claude 응답을 간결하게 표시
# JSON이 아닌 자연어 답변인지 확인
if not response_text.strip().startswith('{'):
content = response_text.strip()
else:
# 만약 JSON이면 파싱해서 표시
try:
data = json.loads(response_text)
if "result" in data:
content = data["result"]
else:
content = response_text
except:
content = response_text
state["current_agent"] = "end"
else:
# 배포 분석: 간단한 메시지만 (Decision agent가 상세 결과 표시)
content = "✅ 분석 완료"
state["current_agent"] = "orchestrator"
state["research_data"] = {
"summary": "정보 수집 완료",
"findings": [{"category": "기본", "data": "클러스터 정보 수집 완료"}],
"findings": [{"category": "분석", "data": response_text}],
"recommendations": []
}
state["messages"].append({
"role": "research",
"content": content
})
state["current_agent"] = "orchestrator"
return state
# 최대 반복 도달
print(f"\n⚠️ 최대 반복 횟수 도달 ({max_iterations})")
content = "✅ 분석 완료\n\n기본 클러스터 정보가 수집되었습니다."
# 요청 유형에 따라 다른 출력
if request_type == "information_query":
# 정보 조회: 수집된 정보를 바탕으로 사용자 친화적인 답변 생성
if tool_outputs:
outputs_text = "\n\n".join(tool_outputs)
# Claude에게 결과 해석 요청
print("\n📝 결과 해석 요청 중...")
interpretation_prompt = f"""수집된 정보를 바탕으로 사용자 질문에 답변해주세요.
**사용자 질문:** {user_message}
**수집된 정보:**
{outputs_text}
위 정보를 바탕으로:
1. 사용자 질문에 직접적으로 답변
2. 한국어로 간결하게 작성
3. 핵심 정보만 포함
4. 기술적 세부사항은 필요시에만 포함
답변:"""
interpretation_response = claude_research.invoke([
HumanMessage(content=interpretation_prompt)
])
content = f"✅ 조회 완료\n\n{interpretation_response.content}"
state["research_data"] = {
"summary": "정보 수집 완료",
"findings": [{"category": "클러스터", "data": "기본 정보 수집 완료"}],
"findings": [{"category": "클러스터 정보", "data": outputs_text}],
"recommendations": []
}
else:
content = "✅ 조회 완료\n\n⚠️ 충분한 정보를 수집하지 못했습니다."
state["research_data"] = {
"summary": "정보 수집 불완전",
"findings": [{"category": "경고", "data": "추가 정보 필요"}],
"recommendations": []
}
state["current_agent"] = "end"
else:
# 배포 분석: 간단한 메시지만 (Decision agent가 상세 결과 표시)
content = "✅ 분석 완료"
if tool_outputs:
outputs_text = "\n\n".join(tool_outputs)
state["research_data"] = {
"summary": "정보 수집 완료",
"findings": [{"category": "클러스터 정보", "data": outputs_text}],
"recommendations": []
}
else:
state["research_data"] = {
"summary": "정보 수집 불완전",
"findings": [{"category": "경고", "data": "추가 정보 필요"}],
"recommendations": []
}
state["current_agent"] = "orchestrator"
state["messages"].append({
"role": "research",
"content": content
})
state["current_agent"] = "orchestrator"
return state

4
services/backend/chainlit.md
View File

@@ -8,7 +8,3 @@ Hi there, Developer! 👋 We're excited to have you on board. Chainlit is a powe
- **Discord Community:** Join our friendly [Chainlit Discord](https://discord.gg/k73SQ3FyUh) to ask questions, share your projects, and connect with other developers! 💬
We can't wait to see what you create with Chainlit! Happy coding! 💻😊
## Welcome screen
To modify the welcome screen, edit the `chainlit.md` file at the root of your project. If you do not want a welcome screen, just leave this file empty.

6
services/backend/tools/bash_tool.py
View File

@@ -2,6 +2,7 @@
Bash 명령어 실행 도구
"""
import subprocess
import shlex
from langchain_core.tools import tool
from typing import Optional
@@ -95,12 +96,13 @@ def execute_host(command: str, timeout: int = 30, use_sudo: bool = False) -> str
# This allows commands to work from SSH initial directory
if use_sudo:
# For sudo commands, run directly with sudo
nsenter_command = f"nsenter -t 1 -m -u -n -i -- sh -c {subprocess.list2cmdline([f'sudo {command}'])}"
# Use shlex.quote to safely quote the command while preserving shell expansion
nsenter_command = f"nsenter -t 1 -m -u -n -i -- sh -c {shlex.quote(f'sudo {command}')}"
else:
# For regular commands, run as ubuntu user
# Use 'su ubuntu -c' (not 'su - ubuntu -c') to preserve current directory
# This matches SSH behavior where you start from the initial directory
nsenter_command = f"nsenter -t 1 -m -u -n -i -- su ubuntu -c {subprocess.list2cmdline([command])}"
nsenter_command = f"nsenter -t 1 -m -u -n -i -- su ubuntu -c {shlex.quote(command)}"
result = subprocess.run(
nsenter_command,