Mercurial Repositories > eric / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2020 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing a check for shell injection.
"""

#
# This is a modified version of the one found in the bandit package.
#
# Original Copyright 2014 Hewlett-Packard Development Company, L.P.
#
# SPDX-License-Identifier: Apache-2.0
#

import ast
import re
import sys

# This regex starts with a windows drive letter (eg C:)
# or one of our path delimeter characters (/, \, .)
fullPathMatchRe = re.compile(r'^(?:[A-Za-z](?=\:)|[\\\/\.])')


def getChecks():
    """
    Public method to get a dictionary with checks handled by this module.
    
    @return dictionary containing checker lists containing checker function and
        list of codes
    @rtype dict
    """
    return {
        "Call": [
            (checkSubprocessPopenWithShell, ("S602",)),
            (checkSubprocessPopenWithoutShell, ("S603",)),
            (checkOtherFunctionWithShell, ("S604",)),
            (checkStartProcessWithShell, ("S605",)),
            (checkStartProcessWithNoShell, ("S606",)),
            (checkStartProcessWithPartialPath, ("S607",)),
        ],
    }


def _defaultValues(key):
    """
    Function to get the default values for a given check key.
    
    @param key key to get default values for
    @type str
    @return list with default values
    @rtype list of str
    """
    if key == "shell_injection_subprocess":
        return [
            'subprocess.Popen',
            'subprocess.call',
            'subprocess.check_call',
            'subprocess.check_output',
            'subprocess.run'
        ]
    elif key == "shell_injection_shell":
        return [
            'os.system',
            'os.popen',
            'os.popen2',
            'os.popen3',
            'os.popen4',
            'popen2.popen2',
            'popen2.popen3',
            'popen2.popen4',
            'popen2.Popen3',
            'popen2.Popen4',
            'commands.getoutput',
            'commands.getstatusoutput'
        ]
    elif key == "shell_injection_noshell":
        return [
            'os.execl',
            'os.execle',
            'os.execlp',
            'os.execlpe',
            'os.execv',
            'os.execve',
            'os.execvp',
            'os.execvpe',
            'os.spawnl',
            'os.spawnle',
            'os.spawnlp',
            'os.spawnlpe',
            'os.spawnv',
            'os.spawnve',
            'os.spawnvp',
            'os.spawnvpe',
            'os.startfile'
        ]
    else:
        return []


def _evaluateShellCall(context):
    """
    Function to determine the severity of a shell call.
    
    @param context context to be inspected
    @type SecurityContext
    @return severity level (L, M or H)
    @rtype str
    """
    noFormatting = isinstance(context.node.args[0], ast.Str)

    if noFormatting:
        return "L"
    else:
        return "H"


def hasShell(context):
    """
    Function to check, if the node of the context contains the shell keyword.
    
    @param context context to be inspected
    @type SecurityContext
    @return tuple containing a flag indicating the presence of the 'shell'
        argument and flag indicating the value of the 'shell' argument
    @rtype tuple of (bool, bool)
    """
    keywords = context.node.keywords
    result = False
    shell = False
    if 'shell' in context.callKeywords:
        shell = True
        for key in keywords:
            if key.arg == 'shell':
                val = key.value
                if isinstance(val, ast.Num):
                    result = bool(val.n)
                elif isinstance(val, ast.List):
                    result = bool(val.elts)
                elif isinstance(val, ast.Dict):
                    result = bool(val.keys)
                elif isinstance(val, ast.Name) and val.id in ['False', 'None']:
                    result = False
                elif (
                    sys.version_info[0] > 2 and
                    isinstance(val, ast.NameConstant)
                ):
                    result = val.value
                else:
                    result = True
    
    return shell, result


def checkSubprocessPopenWithShell(reportError, context, config):
    """
    Function to check for use of popen with shell equals true.
    
    @param reportError function to be used to report errors
    @type func
    @param context security context object
    @type SecurityContext
    @param config dictionary with configuration data
    @type dict
    """
    if config and "shell_injection_subprocess" in config:
        functionNames = config["shell_injection_subprocess"]
    else:
        functionNames = _defaultValues("shell_injection_subprocess")
    
    if context.callFunctionNameQual in functionNames:
        shell, shellValue = hasShell(context)
        if shell and shellValue:
            if len(context.callArgs) > 0:
                sev = _evaluateShellCall(context)
                if sev == "L":
                    reportError(
                        context.getLinenoForCallArg('shell') - 1,
                        context.getOffsetForCallArg('shell'),
                        "S602.L",
                        sev,
                        "H",
                    )
                else:
                    reportError(
                        context.getLinenoForCallArg('shell') - 1,
                        context.getOffsetForCallArg('shell'),
                        "S602.H",
                        sev,
                        "H",
                    )


def checkSubprocessPopenWithoutShell(reportError, context, config):
    """
    Function to check for use of popen without shell equals true.
    
    @param reportError function to be used to report errors
    @type func
    @param context security context object
    @type SecurityContext
    @param config dictionary with configuration data
    @type dict
    """
    if config and "shell_injection_subprocess" in config:
        functionNames = config["shell_injection_subprocess"]
    else:
        functionNames = _defaultValues("shell_injection_subprocess")
    
    if context.callFunctionNameQual in functionNames:
        if not hasShell(context)[0]:
            reportError(
                context.node.lineno - 1,
                context.node.col_offset,
                "S603",
                "L",
                "H",
            )


def checkOtherFunctionWithShell(reportError, context, config):
    """
    Function to check for any function with shell equals true.
    
    @param reportError function to be used to report errors
    @type func
    @param context security context object
    @type SecurityContext
    @param config dictionary with configuration data
    @type dict
    """
    if config and "shell_injection_subprocess" in config:
        functionNames = config["shell_injection_subprocess"]
    else:
        functionNames = _defaultValues("shell_injection_subprocess")
    
    if context.callFunctionNameQual not in functionNames:
        shell, shellValue = hasShell(context)
        if shell and shellValue:
            reportError(
                context.getLinenoForCallArg('shell') - 1,
                context.getOffsetForCallArg('shell'),
                "S604",
                "M",
                "L",
            )


def checkStartProcessWithShell(reportError, context, config):
    """
    Function to check for starting a process with a shell.
    
    @param reportError function to be used to report errors
    @type func
    @param context security context object
    @type SecurityContext
    @param config dictionary with configuration data
    @type dict
    """
    if config and "shell_injection_shell" in config:
        functionNames = config["shell_injection_shell"]
    else:
        functionNames = _defaultValues("shell_injection_shell")
    
    if context.callFunctionNameQual in functionNames:
        if len(context.callArgs) > 0:
            sev = _evaluateShellCall(context)
            if sev == "L":
                reportError(
                    context.node.lineno - 1,
                    context.node.col_offset,
                    "S605.L",
                    sev,
                    "H",
                )
            else:
                reportError(
                    context.node.lineno - 1,
                    context.node.col_offset,
                    "S605.H",
                    sev,
                    "H",
                )


def checkStartProcessWithNoShell(reportError, context, config):
    """
    Function to check for starting a process with no shell.
    
    @param reportError function to be used to report errors
    @type func
    @param context security context object
    @type SecurityContext
    @param config dictionary with configuration data
    @type dict
    """
    if config and "shell_injection_noshell" in config:
        functionNames = config["shell_injection_noshell"]
    else:
        functionNames = _defaultValues("shell_injection_noshell")
    
    if context.callFunctionNameQual in functionNames:
        reportError(
            context.node.lineno - 1,
            context.node.col_offset,
            "S606",
            "L",
            "M",
        )


def checkStartProcessWithPartialPath(reportError, context, config):
    """
    Function to check for starting a process with no shell.
    
    @param reportError function to be used to report errors
    @type func
    @param context security context object
    @type SecurityContext
    @param config dictionary with configuration data
    @type dict
    """
    if config and "shell_injection_subprocess" in config:
        functionNames = config["shell_injection_subprocess"]
    else:
        functionNames = _defaultValues("shell_injection_subprocess")
    
    if config and "shell_injection_shell" in config:
        functionNames += config["shell_injection_shell"]
    else:
        functionNames += _defaultValues("shell_injection_shell")
    
    if config and "shell_injection_noshell" in config:
        functionNames += config["shell_injection_noshell"]
    else:
        functionNames += _defaultValues("shell_injection_noshell")
    
    if len(context.callArgs):
        if context.callFunctionNameQual in functionNames:
            node = context.node.args[0]
            
            # some calls take an arg list, check the first part
            if isinstance(node, ast.List):
                node = node.elts[0]
            
            # make sure the param is a string literal and not a var name
            if isinstance(node, ast.Str) and not fullPathMatchRe.match(node.s):
                reportError(
                    context.node.lineno - 1,
                    context.node.col_offset,
                    "S607",
                    "L",
                    "H",
                )