mysql-connector-python/lib/mysql/connector/aio/connection.py [401:481]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        self,
        filename: str,
        read_timeout: Optional[int] = None,
        write_timeout: Optional[int] = None,
    ) -> OkPacketType:
        """Handle a LOAD DATA INFILE LOCAL request."""
        if self._local_infile_filenames is None:
            self._local_infile_filenames = get_local_infile_filenames(self._query)
            if not self._local_infile_filenames:
                raise InterfaceError(
                    "No `LOCAL INFILE` statements found in the client's request. "
                    "Check your request includes valid `LOCAL INFILE` statements."
                )
        elif not self._local_infile_filenames:
            raise InterfaceError(
                "Got more `LOCAL INFILE` responses than number of `LOCAL INFILE` "
                "statements specified in the client's request. Please, report this "
                "issue to the development team."
            )

        file_name = os.path.abspath(filename)
        file_name_from_request = os.path.abspath(self._local_infile_filenames.popleft())

        # Verify the file location specified by `filename` from client's request exists
        if not os.path.exists(file_name_from_request):
            raise InterfaceError(
                f"Location specified by filename {file_name_from_request} "
                "from client's request does not exist."
            )

        # Verify the file location specified by `filename` from server's response exists
        if not os.path.exists(file_name):
            raise InterfaceError(
                f"Location specified by filename {file_name} from server's "
                "response does not exist."
            )

        # Verify the `filename` specified by server's response matches the one from
        # the client's request.
        try:
            if not os.path.samefile(file_name, file_name_from_request):
                raise InterfaceError(
                    f"Filename {file_name} from the server's response is not the same "
                    f"as filename {file_name_from_request} from the "
                    "client's request."
                )
        except OSError as err:
            raise InterfaceError from err

        if os.path.islink(file_name):
            raise OperationalError("Use of symbolic link is not allowed")
        if not self._allow_local_infile and not self._allow_local_infile_in_path:
            raise DatabaseError(
                "LOAD DATA LOCAL INFILE file request rejected due to "
                "restrictions on access."
            )
        if not self._allow_local_infile and self._allow_local_infile_in_path:
            # Validate filename is inside of allow_local_infile_in_path path.
            infile_path = os.path.abspath(self._allow_local_infile_in_path)
            c_path = None
            try:
                c_path = os.path.commonpath([infile_path, file_name])
            except ValueError as err:
                err_msg = (
                    "{} while loading file `{}` and path `{}` given"
                    " in allow_local_infile_in_path"
                )
                raise InterfaceError(
                    err_msg.format(str(err), file_name, infile_path)
                ) from err

            if c_path != infile_path:
                err_msg = (
                    "The file `{}` is not found in the given "
                    "allow_local_infile_in_path {}"
                )
                raise DatabaseError(err_msg.format(file_name, infile_path))

        try:
            data_file = open(file_name, "rb")  # pylint: disable=consider-using-with
            return self._handle_ok(
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



mysql-connector-python/lib/mysql/connector/connection.py [617:697]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        self,
        filename: str,
        read_timeout: Optional[int] = None,
        write_timeout: Optional[int] = None,
    ) -> OkPacketType:
        """Handle a LOAD DATA INFILE LOCAL request"""
        if self._local_infile_filenames is None:
            self._local_infile_filenames = get_local_infile_filenames(self._query)
            if not self._local_infile_filenames:
                raise InterfaceError(
                    "No `LOCAL INFILE` statements found in the client's request. "
                    "Check your request includes valid `LOCAL INFILE` statements."
                )
        elif not self._local_infile_filenames:
            raise InterfaceError(
                "Got more `LOCAL INFILE` responses than number of `LOCAL INFILE` "
                "statements specified in the client's request. Please, report this "
                "issue to the development team."
            )

        file_name = os.path.abspath(filename)
        file_name_from_request = os.path.abspath(self._local_infile_filenames.popleft())

        # Verify the file location specified by `filename` from client's request exists
        if not os.path.exists(file_name_from_request):
            raise InterfaceError(
                f"Location specified by filename {file_name_from_request} "
                "from client's request does not exist."
            )

        # Verify the file location specified by `filename` from server's response exists
        if not os.path.exists(file_name):
            raise InterfaceError(
                f"Location specified by filename {file_name} from server's "
                "response does not exist."
            )

        # Verify the `filename` specified by server's response matches the one from
        # the client's request.
        try:
            if not os.path.samefile(file_name, file_name_from_request):
                raise InterfaceError(
                    f"Filename {file_name} from the server's response is not the same "
                    f"as filename {file_name_from_request} from the "
                    "client's request."
                )
        except OSError as err:
            raise InterfaceError from err

        if os.path.islink(file_name):
            raise OperationalError("Use of symbolic link is not allowed")
        if not self._allow_local_infile and not self._allow_local_infile_in_path:
            raise DatabaseError(
                "LOAD DATA LOCAL INFILE file request rejected due to "
                "restrictions on access."
            )
        if not self._allow_local_infile and self._allow_local_infile_in_path:
            # validate filename is inside of allow_local_infile_in_path path.
            infile_path = os.path.abspath(self._allow_local_infile_in_path)
            c_path = None
            try:
                c_path = os.path.commonpath([infile_path, file_name])
            except ValueError as err:
                err_msg = (
                    "{} while loading file `{}` and path `{}` given"
                    " in allow_local_infile_in_path"
                )
                raise InterfaceError(
                    err_msg.format(str(err), file_name, infile_path)
                ) from err

            if c_path != infile_path:
                err_msg = (
                    "The file `{}` is not found in the given "
                    "allow_local_infile_in_path {}"
                )
                raise DatabaseError(err_msg.format(file_name, infile_path))

        try:
            data_file = open(file_name, "rb")  # pylint: disable=consider-using-with
            return self._handle_ok(
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -