94 lines
2.6 KiB
Python
94 lines
2.6 KiB
Python
|
"""
|
||
|
|
Native Runtime
|
||
|
|
|
||
|
|
Native binary execution runtime for system executables.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import stat
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Dict, List
|
||
|
|
|
||
|
|
from tools.runtime.base import BaseRuntime
|
||
|
|
|
||
|
|
|
||
|
|
class NativeRuntime(BaseRuntime):
|
||
|
|
"""Native binary execution runtime"""
|
||
|
|
|
||
|
|
def get_name(self) -> str:
|
||
|
|
return "native"
|
||
|
|
|
||
|
|
async def validate(self, entry: str) -> bool:
|
||
|
|
"""Validate native binary exists and is executable"""
|
||
|
|
path = Path(entry)
|
||
|
|
if not path.exists():
|
||
|
|
return False
|
||
|
|
|
||
|
|
# Check if file is executable
|
||
|
|
file_stat = path.stat()
|
||
|
|
executable_bit = stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
|
||
|
|
if not (file_stat.st_mode & executable_bit):
|
||
|
|
return False
|
||
|
|
|
||
|
|
return True
|
||
|
|
|
||
|
|
async def execute(
|
||
|
|
self,
|
||
|
|
entry: str,
|
||
|
|
command: str,
|
||
|
|
parameters: Dict[str, Any],
|
||
|
|
timeout: int,
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
"""Execute a native binary"""
|
||
|
|
try:
|
||
|
|
# Build argument list
|
||
|
|
args = [entry, command] + self._format_args(parameters)
|
||
|
|
|
||
|
|
process = await asyncio.create_subprocess_exec(
|
||
|
|
*args,
|
||
|
|
stdout=asyncio.subprocess.PIPE,
|
||
|
|
stderr=asyncio.subprocess.PIPE,
|
||
|
|
)
|
||
|
|
|
||
|
|
stdout, stderr = await asyncio.wait_for(
|
||
|
|
process.communicate(),
|
||
|
|
timeout=timeout / 1000,
|
||
|
|
)
|
||
|
|
|
||
|
|
if process.returncode != 0:
|
||
|
|
return {
|
||
|
|
"status": "error",
|
||
|
|
"error": stderr.decode() if stderr else f"Exit code: {process.returncode}",
|
||
|
|
}
|
||
|
|
|
||
|
|
return {
|
||
|
|
"status": "success",
|
||
|
|
"result": stdout.decode() if stdout else "",
|
||
|
|
}
|
||
|
|
|
||
|
|
except asyncio.TimeoutError:
|
||
|
|
return {
|
||
|
|
"status": "error",
|
||
|
|
"error": f"Execution timed out after {timeout}ms",
|
||
|
|
}
|
||
|
|
except Exception as e:
|
||
|
|
return {
|
||
|
|
"status": "error",
|
||
|
|
"error": str(e),
|
||
|
|
}
|
||
|
|
|
||
|
|
def _format_args(self, parameters: Dict[str, Any]) -> List[str]:
|
||
|
|
"""Format parameters as command-line arguments"""
|
||
|
|
args: List[str] = []
|
||
|
|
for key, value in parameters.items():
|
||
|
|
# Use --key=value format
|
||
|
|
if isinstance(value, bool):
|
||
|
|
if value:
|
||
|
|
args.append(f"--{key}")
|
||
|
|
elif isinstance(value, (list, tuple)):
|
||
|
|
for item in value:
|
||
|
|
args.extend([f"--{key}", str(item)])
|
||
|
|
else:
|
||
|
|
args.extend([f"--{key}", str(value)])
|
||
|
|
return args
|