"""Bounded static wrapper views. No shell/Python execution and no permission rules. Original command, request, host and trust fields stay intact. Only recognized constant wrappers yield an additional view; unsupported syntax remains untouched. This is a pipeline feature, not evidence of user authorization or a safety proof. """ import ast import base64 import re import shlex VERSION='static-wrappers-v2' FIELDS={'static_command_view','static_execution_chain','static_view_origin'} MAX_TEXT=16384 def _python_source(code): try:tree=ast.parse(code) except (SyntaxError,ValueError,MemoryError,RecursionError):return None aliases={};calls=[] for node in tree.body: if isinstance(node,ast.Import): if calls:return None if any(x.name not in ('os','subprocess') for x in node.names):return None aliases.update({x.asname or x.name:x.name for x in node.names}) elif isinstance(node,ast.ImportFrom) and node.module in ('os','subprocess') and node.level==0: if calls:return None aliases.update({x.asname or x.name:node.module+'.'+x.name for x in node.names}) elif isinstance(node,ast.Expr) and isinstance(node.value,ast.Call):calls.append(node.value) else:return None if len(calls)!=1:return None call=calls[0] if isinstance(call.func,ast.Attribute) and isinstance(call.func.value,ast.Name): name=aliases.get(call.func.value.id,'')+'.'+call.func.attr elif isinstance(call.func,ast.Name):name=aliases.get(call.func.id,'') else:return None if len(call.args)!=1 or any(k.arg is None for k in call.keywords):return None try: arg=ast.literal_eval(call.args[0]);kw={k.arg:ast.literal_eval(k.value) for k in call.keywords} except (ValueError,TypeError,MemoryError,RecursionError):return None if name=='os.system' and not kw and isinstance(arg,str):return arg if name not in ('subprocess.run','subprocess.call','subprocess.check_call','subprocess.check_output','subprocess.Popen'):return None if set(kw)-{'shell','check','capture_output','text','encoding','timeout'}:return None if isinstance(arg,(list,tuple)) and all(isinstance(x,str) for x in arg) and not kw.get('shell'): return shlex.join(arg) if isinstance(arg,str) and kw.get('shell') is True:return arg return None def _unwrap(command): # Quoted shell heredoc: the body is executed by this shell, not by cat. match=re.fullmatch(r"(?:bash|sh)[ \t]+<<[ \t]*(['\"])([A-Za-z_][A-Za-z0-9_]*)\1\n(.*)\n\2(?:\n)?",command,re.S) if match and match.group(2) not in match.group(3).splitlines(): return match.group(3),'quoted shell heredoc' try: lexer=shlex.shlex(command,posix=True,punctuation_chars='|;&<>()') lexer.whitespace_split=True;lexer.commenters='' argv=list(lexer) except ValueError:return None if len(argv)==8 and argv[:2]==['printf','%s'] and argv[3:5]==['|','base64'] and argv[5] in ('--decode','-d') and argv[6]=='|' and argv[7] in ('sh','bash'): try:payload=base64.b64decode(argv[2],validate=True).decode('utf-8') except (ValueError,UnicodeError):return None if payload and len(payload)<=MAX_TEXT:return payload,'constant base64 piped into shell' if len(argv)==3 and argv[0] in ('bash','sh') and argv[1]=='-c' and '$' not in command and '`' not in command: return argv[2],argv[0]+' -c constant source' if len(argv)==3 and argv[0]=='ssh' and re.fullmatch(r'[A-Za-z0-9_.@:-]+',argv[1]) and not argv[1].startswith('-') and '$' not in command and '`' not in command: return argv[2],'ssh target='+argv[1] if len(argv)==3 and argv[0] in ('python','python3') and argv[1]=='-c' and '$' not in command and '`' not in command: payload=_python_source(argv[2]) if payload and len(payload)<=MAX_TEXT:return payload,'constant Python process invocation' return None def normalize_state(state): result={k:v for k,v in state.items() if k not in FIELDS} command=result.get('command') if result.get('tool')!='Bash' or not isinstance(command,str) or len(command)>MAX_TEXT:return result current=command;chain=[] for _ in range(4): unwrapped=_unwrap(current) if not unwrapped:break payload,description=unwrapped if payload==current:break chain.append(description);current=payload if chain: result['static_command_view']=current result['static_execution_chain']=chain result['static_view_origin']=VERSION return result