command_normalizer_v2.py

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