| """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): |
| |
| 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 |
|
|
