WIP: OSCMD2:Add oscmdpatches report for Command Injection analysis

chb/api/FormatStringSpec.py

@@ -110,6 +110,16 @@ def lengthmodifier(self) -> str:
     def conversion(self) -> str:
         return self.tags[3]
 
+    @property
+    def arg_type(self) -> str:
+        if self.conversion == 's':
+            return 'string'
+        elif self.conversion == 'd':
+            return 'int'
+        elif self.conversion == 'f':
+            return 'float'
+        return 'unknown'
+
     @property
     def flags(self) -> str:
         return "".join(chr(i) for i in self.args)

chb/cmdline/chkx

@@ -1334,6 +1334,42 @@ def parse() -> argparse.Namespace:
 
     report_patchcandidates.set_defaults(func=REP.report_patch_candidates)
 
+    # --- report patch candidates
+    report_oscmdcandidates = reportparsers.add_parser("oscmdcandidates")
+    report_oscmdcandidates.add_argument("xname", help="name of executable")
+    report_oscmdcandidates.add_argument(
+        "--output", "-o",
+        help="name of output file (without extension)")
+    report_oscmdcandidates.add_argument(
+        "--json",
+        action="store_true",
+        help="output results in json format")
+    report_oscmdcandidates.add_argument(
+        "--targets",
+        nargs="*",
+        default=['all'],
+        help="list of target library functions to include. If not passed, all "
+             "library functions found will be included.")
+    report_oscmdcandidates.add_argument(
+        "--verbose", "-v",
+        action="store_true",
+        help="print functions examined")
+    report_oscmdcandidates.add_argument(
+        "--loglevel", "-log",
+        choices=UL.LogLevel.options(),
+        default="NONE",
+        help="activate logging with the given level (default to stderr)")
+    report_oscmdcandidates.add_argument(
+        "--logfilename",
+        help="name of file to write log messages")
+    report_oscmdcandidates.add_argument(
+        "--logfilemode",
+        choices=["a", "w"],
+        default="a",
+        help="file mode for log file: append (a, default), or write (w)")
+
+    report_oscmdcandidates.set_defaults(func=REP.report_os_cmd_candidates)
+
     '''
 
     # -- report application calls --

chb/cmdline/reportcmds.py

@@ -49,6 +49,8 @@
 from chb.app.Callgraph import Callgraph
 from chb.app.Instruction import Instruction
 
+from chb.app.XPOPredicate import XPOTrustedOsCmdFmtString, XPOTrustedOsCmdString
+
 from chb.arm.ARMInstruction import ARMInstruction
 
 from chb.bctypes.BCAttrParam import BCAttrParamInt, BCAttrParamStr, BCAttrParamCons
@@ -70,10 +72,11 @@
 
 if TYPE_CHECKING:
     from chb.api.CallTarget import (
-        StubTarget, CallTarget)
+        StubTarget, AppTarget, CallTarget)
     from chb.api.FunctionStub import SOFunction
     from chb.app.AppAccess import AppAccess
     from chb.app.BasicBlock import BasicBlock
+    from chb.app.FnProofObligations import ProofObligation
     from chb.app.Function import Function
     from chb.app.FunctionStackframe import FunctionStackframe
     from chb.app.Instruction import Instruction
@@ -1452,7 +1455,6 @@ def include_target(target: 'CallTarget') -> bool:
             intermediate_attribute = find_function_attribute(app, dstarg_index, fname, instr, pc)
             if intermediate_attribute is None:
                 continue
-
             for inter in intermediate_callgraph[fname]:
                 # For each caller to the intermediate function, lookup the corresponding buffer
                 # and add it to the patch records.
@@ -1520,3 +1522,171 @@ def include_target(target: 'CallTarget') -> bool:
           (len(content['patch-records']), n_calls))
 
     exit(0)
+
+def report_os_cmd_candidates(args: argparse.Namespace) -> NoReturn:
+
+    # arguments
+    xname = args.xname
+    xoutput: Optional[str] = args.output
+    xjson: bool = args.json
+    xverbose: bool = args.verbose
+    xtargets: List[str] = args.targets
+    loglevel: str = args.loglevel
+    logfilename: Optional[str] = args.logfilename
+    logfilemode: str = args.logfilemode
+
+    try:
+        (path, xfile) = UC.get_path_filename(xname)
+        UF.check_analysis_results(path, xfile)
+    except UF.CHBError as e:
+        print(str(e.wrap()))
+        exit(1)
+
+    UC.set_logging(
+        loglevel,
+        path,
+        logfilename=logfilename,
+        mode=logfilemode,
+        msg="report_os_cmd_patch_candidates invoked")
+
+    xinfo = XI.XInfo()
+    xinfo.load(path, xfile)
+
+    app = UC.get_app(path, xfile, xinfo)
+
+    n_calls: int = 0
+    libcalls = LibraryCallCallsites()
+
+    include_all = xtargets == ['all']
+
+    def include_target(target: 'CallTarget') -> bool:
+        if include_all:
+            return True
+        return target.name in xtargets
+
+    def is_cmd_delegation(po: 'ProofObligation') -> bool:
+        if type(po.xpo) == XPOTrustedOsCmdString:
+            return po.status.is_delegated_local and po.status.get_iaddr() != instr.iaddr
+        return False
+
+    # Track every call with a trusted-os-cmd-fmt-string proofobligation as potential
+    # patch sites for command injection vulnerabilities. These correspond to the site
+    # where the format string is constructed. We additionally track all the
+    # trusted-os-cmd-string delegations as potential sites where the actual command
+    # execution occurs, and allows consumers to filter based on those instruction
+    # addresses.
+    os_cmd_construction: List[tuple[str, "Instruction", str]]  = []
+    os_cmd_delegations: Dict[str,List[str]] = defaultdict(list)
+    app_functions: Dict[str, str] = {}
+
+    for (faddr, blocks) in app.call_instructions().items():
+        for (baddr, instrs) in blocks.items():
+            for instr in instrs:
+                n_calls += 1
+                calltgt = instr.call_target
+                if include_target(calltgt):
+                    for po in instr.proofobligations():
+                        if is_cmd_delegation(po):
+                            os_cmd_delegations[po.status.get_iaddr()].append(instr.iaddr)
+                        if type(po.xpo) == XPOTrustedOsCmdFmtString:
+                            os_cmd_construction.append((faddr, instr, calltgt.name))
+                            if calltgt.is_app_target and calltgt.name not in app_functions:
+                                app_functions[calltgt.name] = str(cast('AppTarget', calltgt).address)
+                if calltgt.is_so_target:
+                    libcalls.add_library_callsite(faddr, baddr, instr)
+
+    chklogger.logger.debug("Number of calls: %s", n_calls)
+
+    patchcallsites = libcalls.patch_callsites()
+
+    function_addr = collect_known_fn_addrs(app, patchcallsites)
+    for name, addr in app_functions.items():
+        function_addr[name] = addr
+
+    content: Dict[str, Any] = {}
+    if xjson:
+        xinfodata = xinfo.to_json_result()
+        if xinfodata.is_ok:
+            content["identification"] = xinfodata.content
+        else:
+            write_json_result(xoutput, xinfodata, "patchcandidates")
+
+    patch_records = []
+
+    for faddr, instr, target_function in os_cmd_construction:
+        pc_content: Dict[str, Any] = {}
+        pc_content["annotation"] = instr.annotation
+        pc_content["faddr"] = faddr
+        pc_content["iaddr"] = instr.iaddr
+        # If the command execution and construction are separate patch sites,
+        # we add them as exec-iaddrs to allow for filtering, otherwise we add
+        # the current site as the exec-iaddrs.
+        if instr.iaddr in os_cmd_delegations:
+            pc_content["exec-iaddrs"] = os_cmd_delegations[instr.iaddr]
+        else:
+            pc_content["exec-iaddrs"] = [instr.iaddr]
+        pc_content["target-function"] = target_function
+        fn_args: List[Dict[str, Any]] = []
+        fmt_string, fmt_string_specs = app.function(faddr).formatstrings[instr.iaddr]
+        pc_content["format-string"] = fmt_string
+        found_fmt_arg = False
+        fmt_arg_count = 0
+        for arg in instr.call_arguments:
+            fn_arg: Dict[str, Any] = {"type": "unknown", "role": "unknown", "rep": str(arg)}
+            if arg.is_constant:
+                if arg.is_int_constant:
+                    fn_arg["type"] = "int"
+                elif arg.is_string_reference:
+                    fn_arg["type"] = "string"
+                else:
+                    fn_arg["type"] = "constant"
+            elif arg.is_stack_address:
+                fn = app.function(faddr)
+                stackframe = fn.stackframe
+                argoffset = arg.stack_address_offset()
+                buffersize, sizeorigin = calculate_buffer_size(stackframe, argoffset, instr)
+                fn_arg["type"] = "pointer"
+                fn_arg["max-length"] = buffersize
+                fn_arg["size-origin"] = sizeorigin
+
+            # If we've already found the format string input, treat the remaining arguments
+            # up to the number of format string specifiers as inputs.
+            if found_fmt_arg:
+                fn_arg["role"] = "input"
+                if fmt_arg_count < len(fmt_string_specs.argspecs):
+                    fn_arg["type"] = fmt_string_specs.argspecs[fmt_arg_count].arg_type
+                fmt_arg_count += 1
+            else:
+                fn_arg["role"] = "passthrough"
+                if arg.is_constant and arg.is_string_reference:
+                    fn_arg["type"] = "string"
+                    fn_arg["role"] = "format"
+                    found_fmt_arg = True
+            fn_args.append(fn_arg)
+
+        pc_content["fn_args"] = fn_args
+        patch_records.append(pc_content)
+
+    content["function-addr"] = function_addr
+    content["patch-records"] = patch_records
+    chklogger.logger.debug("Number of patch callsites: %s", len(content['patch-records']))
+
+    if xjson:
+        jcontent = JSONResult("patchcandidates", content, "ok")
+        write_json_result(xoutput, jcontent, "patchcandidates")
+        exit(0)
+
+    for patch_record in content["patch-records"]:
+        print("  " + patch_record['iaddr'] + "  " + patch_record['annotation'])
+        print("  - faddr: %s" % patch_record['faddr'])
+        print("  - exec-iaddrs: %s" % ' '.join([str(i) for i in patch_record['exec-iaddrs']]))
+        print("  - iaddr: %s" % patch_record['iaddr'])
+        print("  - target function: %s" % patch_record['target-function'])
+        print("  - format string: %s" % patch_record['format-string'])
+        print("  - args: %s" % (",".join(['%s(%s)' % (a['role'], a['type']) for a in patch_record['fn_args']])))
+        print("")
+
+    print("Generated %d patch records from %d library calls" %
+          (len(content['patch-records']), n_calls))
+
+    exit(0)