flake8_secure_coding_standard.py

# Copyright 2021 Damien Nguyen
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Main file for the flake8_secure_coding_standard plugin."""

from __future__ import annotations

import ast
import importlib.metadata
import operator
import platform
import stat
from typing import TYPE_CHECKING, Any, AnyStr, ClassVar, Generator

if TYPE_CHECKING:  # pragma: no cover
    import flake8.options.manager

ast_Constant = ast.Constant  # noqa: N816

_use_optparse = tuple(int(s) for s in importlib.metadata.version('flake8').split('.')) < (3, 8, 0)


if _use_optparse:  # pragma: no cover
    import optparse  # noqa: F401 pylint: disable=deprecated-module, unused-import
else:
    import argparse

# ==============================================================================

_DEFAULT_MAX_MODE = 0o755

SCS100 = 'SCS100 use of os.path.abspath() and os.path.relpath() should be avoided in favor of os.path.realpath()'
SCS101 = 'SCS101 `eval()` and `exec()` represent a security risk and should be avoided'
SCS102 = 'SCS102 use of `os.system()` should be avoided'
SCS103 = (
    'SCS103 use of `shell=True` in subprocess functions or use of functions that internally set it should be avoided'
)
SCS104 = 'SCS104 use of `tempfile.mktemp()` should be avoided, prefer `tempfile.mkstemp()`'
SCS105 = (
    'SCS105 use of `yaml.load()` should be avoided, prefer `yaml.safe_load()` or `yaml.load(xxx, Loader=SafeLoader)`'
)
SCS106 = 'SCS106 use of `jsonpickle.decode()` should be avoided'
SCS107 = 'SCS107 debugging code should not be present in production code (e.g. `import pdb`)'
SCS108 = 'SCS108 `assert` statements should not be present in production code'
SCS109 = (
    'SCS109 Use of builtin `open` for writing is discouraged in favor of `os.open` to allow '
    'for setting file permissions'
)
SCS110 = 'SCS110 Use of `os.popen()` should be avoided, as it internally uses `subprocess.Popen` with `shell=True`'
SCS111 = 'SCS111 Use of `shlex.quote()` should be avoided on non-POSIX platforms (such as Windows)'
SCS112 = 'SCS112 Avoid using `os.open` with unsafe permissions (should be {})'
SCS113 = 'SCS113 Avoid using `pickle.load()` and `pickle.loads()`'
SCS114 = 'SCS114 Avoid using `marshal.load()` and `marshal.loads()`'
SCS115 = 'SCS115 Avoid using `shelve.open()`'
SCS116 = 'SCS116 Avoid using `os.mkdir` and `os.makedirs` with unsafe file permissions (should be {})'
SCS117 = 'SCS117 Avoid using `os.mkfifo` with unsafe file permissions (should be {})'
SCS118 = 'SCS118 Avoid using `os.mknod` with unsafe file permissions (should be {})'
SCS119 = 'SCS119 Avoid using `os.chmod` with unsafe file permissions (W ^ X for group and others)'


# ==============================================================================
# Helper functions


def _read_octal_mode_option(name, value, default):  # noqa: C901
    """
    Read an integer or list of integer configuration option.

    Args:
        name (str): Name of option
        value (str): Value of option from the configuration file or on the CLI. Its value can be any of:
            - 'yes', 'y', 'true' (case-insensitive)
                The maximum mode value is then set to self.DEFAULT_MAX_MODE
            - a single octal or decimal integer
                The maximum mode value is then set to that integer value
            - a comma-separated list of integers (octal or decimal)
                The allowed mode values are then those found in the list
            - anything else will count as a falseful value
        default (int,list): Default value for option if set to one of
            ('y', 'yes', 'true') in the configuration file or on the CLI

    Returns:
        A single integer or a (possibly empty) list of integers

    Raises:
        ValueError: if the value of the option is not valid
    """

    def _str_to_int(arg):
        try:
            return int(arg, 8)
        except ValueError:
            return int(arg)

    value = value.lower()
    modes = [mode.strip() for mode in value.split(',')]

    if len(modes) > 1:
        # Lists of allowed modes
        try:
            allowed_modes = [_str_to_int(mode) for mode in modes if mode]
        except ValueError as error:
            msg = f'Unable to convert {modes} elements to integers!'
            raise ValueError(msg) from error
        else:
            if not allowed_modes:
                msg = f'Calculated empty value for `{name}`!'
                raise ValueError(msg)
            return allowed_modes
    elif modes and modes[0]:
        # Single values (ie. max allowed value for mode)
        try:
            return _str_to_int(value)
        except ValueError as error:
            if value in {'y', 'yes', 'true'}:
                return default
            if value in {'n', 'no', 'false'}:
                return None
            msg = f'Invalid value for `{name}`: {value}!'
            raise ValueError(msg) from error
    else:
        msg = f'Invalid value for `{name}`: {value}!'
        raise ValueError(msg)


# ==============================================================================


def _is_posix():
    """Return True if the current system is POSIX-compatible."""
    # NB: we could simply use `os.name` instead of `platform.system()`. However, that solution would be difficult to
    #     test using `mock` as a few modules (like `pytest`) actually use it internally...
    return platform.system() in {'Linux', 'Darwin'}


_is_unix = _is_posix

# ------------------------------------------------------------------------------


def _is_function_call(node: ast.Call, module: AnyStr, function: list[AnyStr] | tuple[AnyStr] | AnyStr) -> bool:
    if not isinstance(function, (list, tuple)):
        function = (function,)
    return (
        isinstance(node.func, ast.Attribute)
        and isinstance(node.func.value, ast.Name)
        and node.func.value.id == module
        and node.func.attr in function
    )


# ==============================================================================


def _is_os_path_call(node: ast.Call) -> bool:
    return (
        isinstance(node.func, ast.Attribute)  # pylint: disable=R0916
        and (
            (isinstance(node.func.value, ast.Name) and node.func.value.id == 'op')
            or (
                isinstance(node.func.value, ast.Attribute)
                and node.func.value.attr == 'path'
                and isinstance(node.func.value.value, ast.Name)
                and node.func.value.value.id == 'os'
            )
        )
        and node.func.attr in {'abspath', 'relpath'}
    )


def _is_builtin_open_for_writing(node: ast.Call) -> bool:
    if isinstance(node.func, ast.Name) and node.func.id == 'open':
        mode = ''
        if len(node.args) > 1:
            if isinstance(node.args[1], ast.Name):
                return True  # variable -> to be on the safe side, flag as inappropriate
            if isinstance(node.args[1], ast_Constant):
                mode = node.args[1].value
            if isinstance(node.args[1], ast.Str):
                mode = node.args[1].s
        else:
            for keyword in node.keywords:
                if keyword.arg == 'mode':
                    if not isinstance(keyword.value, ast_Constant):
                        return True  # variable -> to be on the safe side, flag as inappropriate
                    mode = keyword.value.value
                    break
        if any(m in mode for m in 'awx'):
            # Cover:
            #  * open(..., "w").
            #  * open(..., "wb").
            #  * open(..., "a").
            #  * open(..., "x").
            return True
    return False


def _get_mode_arg(node, args_idx):
    mode = None
    node_intern = None
    if len(node.args) > args_idx:
        node_intern = node.args[args_idx]
    elif node.keywords:
        for keyword in node.keywords:
            if keyword.arg == 'mode':
                node_intern = keyword.value
                break
    if node_intern:
        if isinstance(node_intern, ast_Constant):
            mode = node_intern.value
        elif isinstance(node_intern, ast.Num):  # pragma: no cover
            mode = node_intern.n
    return mode


def _is_allowed_mode(node, allowed_modes, args_idx):
    mode = _get_mode_arg(node, args_idx=args_idx)
    if mode is not None and allowed_modes:
        return mode in allowed_modes

    # NB: default to True in all other cases
    return True


def _is_shell_true_call(node: ast.Call) -> bool:
    _n_args_max = 8
    if not (isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name)):
        return False

    # subprocess module
    if node.func.value.id in {'subprocess', 'sp'}:
        if node.func.attr in {'call', 'check_call', 'check_output', 'Popen', 'run'}:
            for keyword in node.keywords:
                if keyword.arg == 'shell' and isinstance(keyword.value, ast_Constant) and bool(keyword.value.value):
                    return True
            if (
                len(node.args) > _n_args_max
                and isinstance(node.args[_n_args_max], ast_Constant)
                and bool(node.args[_n_args_max].value)
            ):
                return True
        if node.func.attr in {'getoutput', 'getstatusoutput'}:
            return True

    # asyncio module
    return (node.func.value.id == 'asyncio' and node.func.attr == 'create_subprocess_shell') or (
        node.func.value.id == 'loop' and node.func.attr == 'subprocess_shell'
    )


def _is_pdb_call(node: ast.Call) -> bool:
    if isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name) and node.func.value.id == 'pdb':
        # Cover:
        #  * pdb.func().
        return True

    # Cover:
    # * Pdb().
    return isinstance(node.func, ast.Name) and node.func.id == 'Pdb'


def _is_mktemp_call(node: ast.Call) -> bool:
    if isinstance(node.func, ast.Attribute) and node.func.attr == 'mktemp':
        # Cover:
        #  * tempfile.mktemp().
        #  * xxxx.mktemp().
        return True

    # Cover:
    #  * mktemp().
    return isinstance(node.func, ast.Name) and node.func.id == 'mktemp'


def _is_yaml_unsafe_call(node: ast.Call) -> bool:
    _n_args_max = 2
    _safe_loaders = ('BaseLoader', 'SafeLoader')
    _unsafe_loaders = ('Loader', 'UnsafeLoader', 'FullLoader')
    if isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name) and node.func.value.id == 'yaml':
        if node.func.attr in {'unsafe_load', 'full_load'}:
            # Cover:
            #  * yaml.full_load().
            #  * yaml.unsafe_load().
            return True
        if node.func.attr == 'load':
            for keyword in node.keywords:
                if keyword.arg == 'Loader' and isinstance(keyword.value, ast.Name):
                    if keyword.value.id in _unsafe_loaders:
                        # Cover:
                        #  * yaml.load(x, Loader=Loader).
                        #  * yaml.load(x, Loader=UnsafeLoader).
                        #  * yaml.load(x, Loader=FullLoader).
                        return True
                    if keyword.value.id in _safe_loaders:
                        # Cover:
                        #  * yaml.load(x, Loader=BaseLoader).
                        #  * yaml.load(x, Loader=SafeLoader).
                        return False

            if (
                len(node.args) < _n_args_max  # noqa: PLR0916
                or (isinstance(node.args[1], ast.Name) and node.args[1].id in _unsafe_loaders)
                or (
                    isinstance(node.args[1], ast.Attribute)
                    and node.args[1].value.id == 'yaml'
                    and node.args[1].attr in _unsafe_loaders
                )
            ):
                # Cover:
                #  * yaml.load(x).
                #  * yaml.load(x, Loader).
                #  * yaml.load(x, UnsafeLoader).
                #  * yaml.load(x, FullLoader).
                #  * yaml.load(x, yaml.Loader).
                #  * yaml.load(x, yaml.UnsafeLoader).
                #  * yaml.load(x, yaml.FullLoader).
                return True
    # Cover:
    #  * unsafe_load(...).
    #  * full_load(...).
    return isinstance(node.func, ast.Name) and node.func.id in {'unsafe_load', 'full_load'}


# ==============================================================================

_unop = {ast.USub: operator.neg, ast.Not: operator.not_, ast.Invert: operator.inv}
_binop = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.FloorDiv: operator.floordiv,
    ast.Mod: operator.mod,
    ast.BitXor: operator.xor,
    ast.BitOr: operator.or_,
    ast.BitAnd: operator.and_,
}
_chmod_known_mode_values = (
    'S_ISUID',
    'S_ISGID',
    'S_ENFMT',
    'S_ISVTX',
    'S_IREAD',
    'S_IWRITE',
    'S_IEXEC',
    'S_IRWXU',
    'S_IRUSR',
    'S_IWUSR',
    'S_IXUSR',
    'S_IRWXG',
    'S_IRGRP',
    'S_IWGRP',
    'S_IXGRP',
    'S_IRWXO',
    'S_IROTH',
    'S_IWOTH',
    'S_IXOTH',
)


def _chmod_get_mode(node):
    """
    Extract the mode constant of a node.

    Args:
        node: an AST node

    Raises:
        ValueError: if a node is encountered that cannot be processed
    """
    if isinstance(node, ast.Name) and node.id in _chmod_known_mode_values:
        return getattr(stat, node.id)
    if (
        isinstance(node, ast.Attribute)
        and isinstance(node.value, ast.Name)
        and node.attr in _chmod_known_mode_values
        and node.value.id == 'stat'
    ):
        return getattr(stat, node.attr)
    if isinstance(node, ast.UnaryOp):
        return _unop[type(node.op)](_chmod_get_mode(node.operand))
    if isinstance(node, ast.BinOp):
        return _binop[type(node.op)](_chmod_get_mode(node.left), _chmod_get_mode(node.right))

    msg = f'Do not know how to process node: {ast.dump(node)}'
    raise ValueError(msg)


def _chmod_has_wx_for_go(node):
    if platform.system() == 'Windows':
        # On Windows, only stat.S_IREAD and stat.S_IWRITE can be used, all other bits are ignored
        return False

    try:
        modes = None
        if len(node.args) > 1:
            modes = _chmod_get_mode(node.args[1])
        elif node.keywords:
            for keyword in node.keywords:
                if keyword.arg == 'mode':
                    modes = _chmod_get_mode(keyword.value)
                    break
    except ValueError:
        return False
    else:
        if modes is None:
            # NB: this would be from invalid code such as `os.chmod("file.txt")`
            msg = 'Unable to extract `mode` argument from function call!'
            raise RuntimeError(msg)
        # pylint: disable=no-member
        return bool(modes & (stat.S_IWGRP | stat.S_IXGRP | stat.S_IWOTH | stat.S_IXOTH))


# ==============================================================================


class Visitor(ast.NodeVisitor):
    """AST visitor class for the plugin."""

    os_mkdir_modes_allowed: ClassVar[list[int]] = []
    os_mkdir_modes_msg_arg: ClassVar[str] = ''
    os_mkfifo_modes_allowed: ClassVar[list[int]] = []
    os_mkfifo_modes_msg_arg: ClassVar[str] = ''
    os_mknod_modes_allowed: ClassVar[list[int]] = []
    os_mknod_modes_msg_arg: ClassVar[str] = ''
    os_open_modes_allowed: ClassVar[list[int]] = []
    os_open_modes_msg_arg: ClassVar[str] = ''

    mode_msg_map: ClassVar[dict[str, str]] = {
        SCS112: 'open',
        SCS116: 'mkdir',
        SCS117: 'mkfifo',
        SCS118: 'mknod',
    }

    def _format_mode_msg(self, msg_id):
        """Format a mode message."""
        return self.__class__.format_mode_msg(msg_id)

    @classmethod
    def format_mode_msg(cls, msg_id):
        """Format a mode message."""
        return msg_id.format(getattr(cls, f'os_{cls.mode_msg_map[msg_id]}_modes_msg_arg'))

    def __init__(self) -> None:
        """Initialize a Visitor object."""
        self.errors: list[tuple[int, int, str]] = []
        self._from_imports: dict[str, str] = {}

    def visit_Call(self, node: ast.Call) -> None:  # noqa: C901, PLR0912
        """Visitor method called for ast.Call nodes."""
        if _is_pdb_call(node):
            self.errors.append((node.lineno, node.col_offset, SCS107))
        elif _is_mktemp_call(node):
            self.errors.append((node.lineno, node.col_offset, SCS104))
        elif _is_yaml_unsafe_call(node):
            self.errors.append((node.lineno, node.col_offset, SCS105))
        elif _is_function_call(node, module='jsonpickle', function='decode'):
            self.errors.append((node.lineno, node.col_offset, SCS106))
        elif _is_function_call(node, module='os', function='system'):
            self.errors.append((node.lineno, node.col_offset, SCS102))
        elif _is_os_path_call(node):
            self.errors.append((node.lineno, node.col_offset, SCS100))
        elif _is_function_call(node, module='os', function='popen'):
            self.errors.append((node.lineno, node.col_offset, SCS110))
        elif _is_shell_true_call(node):
            self.errors.append((node.lineno, node.col_offset, SCS103))
        elif _is_builtin_open_for_writing(node):
            self.errors.append((node.lineno, node.col_offset, SCS109))
        elif isinstance(node.func, ast.Name) and (node.func.id in {'eval', 'exec'}):
            self.errors.append((node.lineno, node.col_offset, SCS101))
        elif not _is_posix() and _is_function_call(node, module='shlex', function='quote'):
            self.errors.append((node.lineno, node.col_offset, SCS111))
        elif (
            _is_function_call(node, module='os', function='open')
            and self.os_open_modes_allowed
            and not _is_allowed_mode(node, self.os_open_modes_allowed, args_idx=2)
        ):
            self.errors.append((node.lineno, node.col_offset, self._format_mode_msg(SCS112)))
        elif _is_function_call(node, module='pickle', function=('load', 'loads')):
            self.errors.append((node.lineno, node.col_offset, SCS113))
        elif _is_function_call(node, module='marshal', function=('load', 'loads')):
            self.errors.append((node.lineno, node.col_offset, SCS114))
        elif _is_function_call(node, module='shelve', function='open'):
            self.errors.append((node.lineno, node.col_offset, SCS115))
        elif _is_function_call(node, module='os', function='chmod') and _chmod_has_wx_for_go(node):
            self.errors.append((node.lineno, node.col_offset, SCS119))
        elif _is_unix():
            if (
                _is_function_call(node, module='os', function=('mkdir', 'makedirs'))
                and self.os_mkdir_modes_allowed
                and not _is_allowed_mode(node, self.os_mkdir_modes_allowed, args_idx=1)
            ):
                self.errors.append((node.lineno, node.col_offset, self._format_mode_msg(SCS116)))
            elif (
                _is_function_call(node, module='os', function='mkfifo')
                and self.os_mkfifo_modes_allowed
                and not _is_allowed_mode(node, self.os_mkfifo_modes_allowed, args_idx=1)
            ):
                self.errors.append((node.lineno, node.col_offset, self._format_mode_msg(SCS117)))
            elif (
                _is_function_call(node, module='os', function='mknod')
                and self.os_mknod_modes_allowed
                and not _is_allowed_mode(node, self.os_mknod_modes_allowed, args_idx=1)
            ):
                self.errors.append((node.lineno, node.col_offset, self._format_mode_msg(SCS118)))

        self.generic_visit(node)

    def visit_Import(self, node: ast.Import) -> None:
        """Visitor method called for ast.Import nodes."""
        for alias in node.names:
            if alias.name == 'pdb':
                # Cover:
                #  * import pdb.
                #  * import pdb as xxx.
                self.errors.append((node.lineno, node.col_offset, SCS107))

        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:  # noqa: C901
        """Visitor method called for ast.ImportFrom nodes."""
        for alias in node.names:
            if (node.module is None and alias.name == 'pdb') or node.module == 'pdb':
                # Cover:
                #  * from pdb import xxx.
                self.errors.append((node.lineno, node.col_offset, SCS107))
            elif node.module == 'tempfile' and alias.name == 'mktemp':
                # Cover:
                #  * from tempfile import mktemp.
                self.errors.append((node.lineno, node.col_offset, SCS104))
            elif node.module in {'os.path', 'op'} and alias.name in {'relpath', 'abspath'}:
                # Cover:
                #  * from os.path import relpath, abspath.
                #  * import os.path as op; from op import relpath, abspath.
                self.errors.append((node.lineno, node.col_offset, SCS100))
            elif (node.module == 'subprocess' and alias.name in {'getoutput', 'getstatusoutput'}) or (
                node.module == 'asyncio' and alias.name == 'create_subprocess_shell'
            ):
                # Cover:
                # * from subprocess import getoutput.
                # * from subprocess import getstatusoutput.
                # * from asyncio import create_subprocess_shell.
                self.errors.append((node.lineno, node.col_offset, SCS103))
            elif node.module == 'os' and alias.name == 'system':
                # Cover:
                # * from os import system.
                self.errors.append((node.lineno, node.col_offset, SCS102))
            elif node.module == 'os' and alias.name == 'popen':
                # Cover:
                # * from os import popen.
                self.errors.append((node.lineno, node.col_offset, SCS110))
            elif not _is_posix() and node.module == 'shlex' and alias.name == 'quote':
                # Cover:
                # * from shlex import quote.
                # * from shlex import quote as quoted.
                self.errors.append((node.lineno, node.col_offset, SCS111))
            elif node.module == 'pickle' and alias.name in {'load', 'loads'}:
                # Cover:
                # * from pickle import load.
                # * from pickle import loads as load.
                self.errors.append((node.lineno, node.col_offset, SCS113))
            elif node.module == 'marshal' and alias.name in {'load', 'loads'}:
                # Cover:
                # * from marshal import load.
                # * from marshal import loads as load.
                self.errors.append((node.lineno, node.col_offset, SCS114))
            elif node.module == 'shelve' and alias.name == 'open':
                # Cover:
                # * from shelve import open.
                self.errors.append((node.lineno, node.col_offset, SCS115))
        self.generic_visit(node)

    def visit_With(self, node: ast.With) -> None:
        """Visitor method called for ast.With nodes."""
        for item in node.items:
            if isinstance(item.context_expr, ast.Call):
                if _is_builtin_open_for_writing(item.context_expr):
                    self.errors.append((node.lineno, node.col_offset, SCS109))
                elif _is_function_call(item.context_expr, module='os', function='open') and not _is_allowed_mode(
                    item.context_expr, self.os_open_modes_allowed, args_idx=2
                ):
                    self.errors.append((node.lineno, node.col_offset, self._format_mode_msg(SCS112)))
                elif _is_function_call(item.context_expr, module='shelve', function='open'):
                    self.errors.append((node.lineno, node.col_offset, SCS115))

    def visit_Assert(self, node: ast.Assert) -> None:
        """Visitor method called for ast.Assert nodes."""
        self.errors.append((node.lineno, node.col_offset, SCS108))
        self.generic_visit(node)


class Plugin:  # pylint: disable=R0903
    """Plugin class."""

    name = __name__
    version = importlib.metadata.version(__name__)

    def __init__(self, tree: ast.AST):
        """Initialize a Plugin object."""
        self._tree = tree

    @classmethod
    def add_options(cls: type[Plugin], option_manager: flake8.options.manager.OptionManager) -> None:
        """Add command line options."""
        options_data: ClassVar[dict[str, dict[str, str | bool]]] = (
            (
                '--os-mkdir-mode',
                {
                    'type': str,
                    'parse_from_config': True,
                    'default': False,
                    'dest': 'os_mkdir_mode',
                    'help': "If provided, configure how 'mode' parameter of the os.mkdir() function are handled",
                },
            ),
            (
                '--os-mkfifo-mode',
                {
                    'type': str,
                    'parse_from_config': True,
                    'default': False,
                    'dest': 'os_mkfifo_mode',
                    'help': "If provided, configure how 'mode' parameter of the os.mkfifo() function are handled",
                },
            ),
            (
                '--os-mknod-mode',
                {
                    'type': str,
                    'parse_from_config': True,
                    'default': False,
                    'dest': 'os_mknod_mode',
                    'help': "If provided, configure how 'mode' parameter of the os.mknod() function are handled",
                },
            ),
            (
                '--os-open-mode',
                {
                    'type': str,
                    'parse_from_config': True,
                    'default': False,
                    'dest': 'os_open_mode',
                    'help': "If provided, configure how 'mode' parameter of the os.open() function are handled",
                },
            ),
        )

        if _use_optparse:  # pragma: no cover
            cls.add_options_optparse(option_manager, options_data)
        else:
            cls.add_options_argparse(option_manager, options_data)

    @classmethod
    def add_options_optparse(
        cls: type[Plugin], option_manager: flake8.options.manager.OptionManager, options_data: tuple
    ) -> None:  # pragma: no cover
        """Add command line options using optparse."""

        def octal_mode_option_callback(option, _, value, parser):
            """Octal mode option callback."""
            setattr(parser.values, f'{option.dest}', _read_octal_mode_option(option.dest, value, _DEFAULT_MAX_MODE))

        callback = {'action': 'callback', 'callback': octal_mode_option_callback}
        for opt_str, kwargs in options_data:
            option_manager.add_option(opt_str, **{**kwargs, **callback})

    @classmethod
    def add_options_argparse(
        cls: type[Plugin], option_manager: flake8.options.manager.OptionManager, options_data: tuple
    ) -> None:
        """Add command line options using argparse."""

        class OctalModeAction(argparse.Action):
            """Action class for octal mode options."""

            def __call__(self, parser, namespace, values, option_string=None):  # noqa: ARG002
                setattr(namespace, self.dest, _read_octal_mode_option(self.dest, values, _DEFAULT_MAX_MODE))

        action = {'action': OctalModeAction}
        for opt_str, kwargs in options_data:
            option_manager.add_option(opt_str, **{**kwargs, **action})

    @classmethod
    def parse_options(cls: type[Plugin], options: argparse.Namespace) -> None:
        """Parse command line options."""

        def _set_mode_option(name, modes):
            if isinstance(modes, int) and modes > 0:
                setattr(Visitor, f'os_{name}_modes_allowed', list(range(modes + 1)))
                setattr(Visitor, f'os_{name}_modes_msg_arg', f'0 < mode < {oct(modes)}')
            elif modes:
                setattr(Visitor, f'os_{name}_modes_allowed', modes)
                setattr(Visitor, f'os_{name}_modes_msg_arg', f'mode in {[oct(mode) for mode in modes]}')
            else:
                getattr(Visitor, f'os_{name}_modes_allowed').clear()

        _set_mode_option('mkdir', options.os_mkdir_mode)
        _set_mode_option('mkfifo', options.os_mkfifo_mode)
        _set_mode_option('mknod', options.os_mknod_mode)
        _set_mode_option('open', options.os_open_mode)

    def run(self) -> Generator[tuple[int, int, str, type[Any]], None, None]:
        """Entry point for flake8."""
        visitor = Visitor()
        visitor.visit(self._tree)

        for line, col, msg in visitor.errors:
            yield line, col, msg, type(self)