rhodecode/lib/vcs/nodes.py

# Copyright (C) 2014-2024 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/

"""
Module holding everything related to vcs nodes, with vcs2 architecture.
"""

import functools
import os
import stat

from zope.cachedescriptors.property import Lazy as LazyProperty

from rhodecode.config.conf import LANGUAGES_EXTENSIONS_MAP
from rhodecode.lib.str_utils import safe_str, safe_bytes
from rhodecode.lib.hash_utils import md5
from rhodecode.lib.vcs import path as vcspath
from rhodecode.lib.vcs.backends.base import EmptyCommit
from rhodecode.lib.vcs.conf.mtypes import get_mimetypes_db
from rhodecode.lib.vcs.exceptions import NodeError
from rhodecode.lib.vcs_common import NodeKind, FILEMODE_DEFAULT

LARGEFILE_PREFIX = ".hglf"


class NodeState:
    ADDED = "added"
    CHANGED = "changed"
    NOT_CHANGED = "not changed"
    REMOVED = "removed"


# TODO: not sure if that should be bytes or str ?
# most probably bytes because content should be bytes and we check it
BIN_BYTE_MARKER = b"\0"



@functools.total_ordering
class Node(object):
    """
    Simplest class representing file or directory on repository.  SCM backends
    should use ``FileNode`` and ``DirNode`` subclasses rather than ``Node``
    directly.

    Node's ``path`` cannot start with slash as we operate on *relative* paths
    only. Moreover, every single node is identified by the ``path`` attribute,
    so it cannot end with slash, too. Otherwise, path could lead to mistakes.
    """

    # RTLO marker allows swapping text, and certain
    # security attacks could be used with this
    RTLO_MARKER = "\u202e"

    commit = None

    def __init__(self, path: bytes, kind):
        self._validate_path(path)  # can throw exception if path is invalid

        self.bytes_path: bytes = path.rstrip(b"/")  # store for mixed encoding, and raw version
        self.str_path: str = safe_str(self.bytes_path)  # we store paths as str
        self.path: str = self.str_path

        if self.bytes_path == b"" and kind != NodeKind.DIR:
            raise NodeError("Only DirNode and its subclasses may be initialized with empty path")
        self.kind = kind

        if self.is_root() and not self.is_dir():
            raise NodeError("Root node cannot be FILE kind")

    def __eq__(self, other):
        if type(self) is not type(other):
            return False
        for attr in ["name", "path", "kind"]:
            if getattr(self, attr) != getattr(other, attr):
                return False
        if self.is_file():
            # FileNode compare, we need to fallback to content compare
            return None
        else:
            # For DirNode's check without entering each dir
            self_nodes_paths = list(sorted(n.path for n in self.nodes))
            other_nodes_paths = list(sorted(n.path for n in self.nodes))
            if self_nodes_paths != other_nodes_paths:
                return False
        return True

    def __lt__(self, other):
        if self.kind < other.kind:
            return True
        if self.kind > other.kind:
            return False
        if self.path < other.path:
            return True
        if self.path > other.path:
            return False

    def __repr__(self):
        maybe_path = getattr(self, "path", "UNKNOWN_PATH")
        return f"<{self.__class__.__name__} {maybe_path!r}>"

    def __str__(self):
        return self.name

    def _validate_path(self, path: bytes):
        self._assert_bytes(path)

        if path.startswith(b"/"):
            raise NodeError(
                f"Cannot initialize Node objects with slash at "
                f"the beginning as only relative paths are supported. "
                f"Got {path}"
            )

    @classmethod
    def _assert_bytes(cls, value):
        if not isinstance(value, bytes):
            raise TypeError(f"Bytes required as input, got {type(value)} of {value}.")

    @LazyProperty
    def parent(self):
        parent_path: bytes = self.get_parent_path()
        if parent_path:
            if self.commit:
                return self.commit.get_node(parent_path)
            return DirNode(parent_path)
        return None

    @LazyProperty
    def has_rtlo(self):
        """Detects if a path has right-to-left-override marker"""
        return self.RTLO_MARKER in self.str_path

    @LazyProperty
    def dir_path(self):
        """
        Returns name of the directory from full path of this vcs node. Empty
        string is returned if there's no directory in the path
        """
        _parts = self.path.rstrip("/").rsplit("/", 1)
        if len(_parts) == 2:
            return _parts[0]
        return ""

    @LazyProperty
    def name(self):
        """
        Returns name of the node so if its path
        then only last part is returned.
        """
        return self.str_path.rstrip("/").split("/")[-1]

    @property
    def kind(self):
        return self._kind

    @kind.setter
    def kind(self, kind):
        if hasattr(self, "_kind"):
            raise NodeError("Cannot change node's kind")
        else:
            self._kind = kind
            # Post setter check (path's trailing slash)
            if self.str_path.endswith("/"):
                raise NodeError("Node's path cannot end with slash")

    def get_parent_path(self) -> bytes:
        """
        Returns node's parent path or empty string if node is root.
        """
        if self.is_root():
            return b""
        str_path = vcspath.dirname(self.bytes_path.rstrip(b"/")) + b"/"

        return safe_bytes(str_path)

    def is_file(self):
        """
        Returns ``True`` if node's kind is ``NodeKind.FILE``, ``False``
        otherwise.
        """
        return self.kind == NodeKind.FILE

    def is_dir(self):
        """
        Returns ``True`` if node's kind is ``NodeKind.DIR``, ``False``
        otherwise.
        """
        return self.kind == NodeKind.DIR

    def is_root(self):
        """
        Returns ``True`` if node is a root node and ``False`` otherwise.
        """
        return self.kind == NodeKind.DIR and self.path == ""

    def is_submodule(self):
        """
        Returns ``True`` if node's kind is ``NodeKind.SUBMODULE``, ``False``
        otherwise.
        """
        return self.kind == NodeKind.SUBMODULE

    def is_largefile(self):
        """
        Returns ``True`` if node's kind is ``NodeKind.LARGEFILE``, ``False``
        otherwise
        """
        return self.kind == NodeKind.LARGE_FILE

    def is_link(self):
        if self.commit:
            return self.commit.is_link(self.bytes_path)
        return False


class FileNode(Node):
    """
    Class representing file nodes.

    :attribute: path: path to the node, relative to repository's root
    :attribute: content: if given arbitrary sets content of the file
    :attribute: commit: if given, first time content is accessed, callback
    :attribute: mode: stat mode for a node. Default is `FILEMODE_DEFAULT`.
    """

    _filter_pre_load = []

    def __init__(self, path: bytes, content: bytes | None = None, commit=None, mode=None, pre_load=None, pre_load_data=None):
        """
        Only one of ``content`` and ``commit`` may be given. Passing both
        would raise ``NodeError`` exception.

        :param path: relative path to the node
        :param content: content may be passed to constructor
        :param commit: if given, will use it to lazily fetch content
        :param mode: ST_MODE (i.e. 0100644)
        """
        if content and commit:
            raise NodeError("Cannot use both content and commit")

        super().__init__(path, kind=NodeKind.FILE)

        self.commit = commit
        if content and not isinstance(content, bytes):
            # File content is one thing that inherently must be bytes
            # we support passing str too, and convert the content
            content = safe_bytes(content)
        self._content = content
        self._mode = mode or FILEMODE_DEFAULT
        if pre_load_data:
            self._store_pre_load(pre_load_data)
        else:
            self._set_bulk_properties(pre_load)

    def __eq__(self, other):
        eq = super().__eq__(other)
        if eq is not None:
            return eq
        return self.content == other.content

    def __hash__(self):
        raw_id = getattr(self.commit, "raw_id", "")
        return hash((self.path, raw_id))

    def __lt__(self, other):
        lt = super().__lt__(other)
        if lt is not None:
            return lt
        return self.content < other.content

    def __repr__(self):
        short_id = getattr(self.commit, "short_id", "")
        return f"<{self.__class__.__name__} path={self.str_path!r}, short_id={short_id}>"

    def _set_bulk_properties(self, pre_load):
        if not pre_load:
            return
        pre_load = [entry for entry in pre_load if entry not in self._filter_pre_load]
        if not pre_load:
            return

        remote = self.commit.get_remote()
        result = remote.bulk_file_request(self.commit.raw_id, self.bytes_path, pre_load)

        self._store_pre_load(result.items())

    def _store_pre_load(self, pre_load_data):
        for attr, value in pre_load_data:
            if attr == "flags":
                self.__dict__["mode"] = safe_str(value)
            elif attr == "size":
                self.__dict__["size"] = value
            elif attr == "data":
                self.__dict__["_content"] = value
            elif attr == "is_binary":
                self.__dict__["is_binary"] = value
            elif attr == "md5":
                self.__dict__["md5"] = value
            else:
                raise ValueError(f"Unsupported attr in bulk_property: {attr}")

    @LazyProperty
    def mode(self):
        """
        Returns lazily mode of the FileNode. If `commit` is not set, would
        use value given at initialization or `FILEMODE_DEFAULT` (default).
        """
        if self.commit:
            mode = self.commit.get_file_mode(self.bytes_path)
        else:
            mode = self._mode
        return mode

    @LazyProperty
    def raw_bytes(self) -> bytes:
        """
        Returns lazily the raw bytes of the FileNode.
        """
        if self.commit:
            if self._content is None:
                self._content = self.commit.get_file_content(self.bytes_path)
            content = self._content
        else:
            content = self._content
        return content

    def content_uncached(self):
        """
        Returns lazily content of the FileNode.
        """
        if self.commit:
            content = self.commit.get_file_content(self.bytes_path)
        else:
            content = self._content
        return content

    def stream_bytes(self):
        """
        Returns an iterator that will stream the content of the file directly from
        vcsserver without loading it to memory.
        """
        if self.commit:
            return self.commit.get_file_content_streamed(self.bytes_path)
        raise NodeError("Cannot retrieve stream_bytes without related commit attribute")

    def metadata_uncached(self):
        """
        Returns md5, binary flag of the file node, without any cache usage.
        """

        content = self.content_uncached()

        is_binary = bool(content and BIN_BYTE_MARKER in content)
        size = 0
        if content:
            size = len(content)

        return is_binary, md5(content), size, content

    @LazyProperty
    def content(self) -> bytes:
        """
        Returns lazily content of the FileNode.
        """
        content = self.raw_bytes
        if content and not isinstance(content, bytes):
            raise ValueError(f"Content is of type {type(content)} instead of bytes")
        return content

    @LazyProperty
    def str_content(self) -> str:
        return safe_str(self.raw_bytes)

    @LazyProperty
    def size(self):
        if self.commit:
            return self.commit.get_file_size(self.bytes_path)
        raise NodeError("Cannot retrieve size of the file without related commit attribute")

    @LazyProperty
    def message(self):
        if self.commit:
            return self.last_commit.message
        raise NodeError("Cannot retrieve message of the file without related " "commit attribute")

    @LazyProperty
    def last_commit(self):
        if self.commit:
            pre_load = ["author", "date", "message", "parents"]
            return self.commit.get_path_commit(self.bytes_path, pre_load=pre_load)
        raise NodeError("Cannot retrieve last commit of the file without related commit attribute")

    def get_mimetype(self):
        """
        Mimetype is calculated based on the file's content. If ``_mimetype``
        attribute is available, it will be returned (backends which store
        mimetypes or can easily recognize them, should set this private
        attribute to indicate that type should *NOT* be calculated).
        """

        if hasattr(self, "_mimetype"):
            if isinstance(self._mimetype, (tuple, list)) and len(self._mimetype) == 2:
                return self._mimetype
            else:
                raise NodeError("given _mimetype attribute must be an 2 element list or tuple")

        db = get_mimetypes_db()
        mtype, encoding = db.guess_type(self.name)

        if mtype is None:
            if not self.is_largefile() and self.is_binary:
                mtype = "application/octet-stream"
                encoding = None
            else:
                mtype = "text/plain"
                encoding = None

                # try with pygments
                try:
                    from pygments.lexers import get_lexer_for_filename

                    mt = get_lexer_for_filename(self.name).mimetypes
                except Exception:
                    mt = None

                if mt:
                    mtype = mt[0]

        return mtype, encoding

    @LazyProperty
    def mimetype(self):
        """
        Wrapper around full mimetype info. It returns only type of fetched
        mimetype without the encoding part. use get_mimetype function to fetch
        full set of (type,encoding)
        """
        return self.get_mimetype()[0]

    @LazyProperty
    def mimetype_main(self):
        return self.mimetype.split("/")[0]

    @classmethod
    def get_lexer(cls, filename, content=None):
        from pygments import lexers

        extension = filename.split(".")[-1]
        lexer = None

        try:
            lexer = lexers.guess_lexer_for_filename(filename, content, stripnl=False)
        except lexers.ClassNotFound:
            pass

        # try our EXTENSION_MAP
        if not lexer:
            try:
                lexer_class = LANGUAGES_EXTENSIONS_MAP.get(extension)
                if lexer_class:
                    lexer = lexers.get_lexer_by_name(lexer_class[0])
            except lexers.ClassNotFound:
                pass

        if not lexer:
            lexer = lexers.TextLexer(stripnl=False)

        return lexer

    @LazyProperty
    def lexer(self):
        """
        Returns pygment's lexer class. Would try to guess lexer taking file's
        content, name and mimetype.
        """
        # TODO: this is more proper, but super heavy on investigating the type based on the content
        # self.get_lexer(self.name, self.content)

        return self.get_lexer(self.name)

    @LazyProperty
    def lexer_alias(self):
        """
        Returns first alias of the lexer guessed for this file.
        """
        return self.lexer.aliases[0]

    @LazyProperty
    def history(self):
        """
        Returns a list of commit for this file in which the file was changed
        """
        if self.commit is None:
            raise NodeError("Unable to get commit for this FileNode")
        return self.commit.get_path_history(self.bytes_path)

    @LazyProperty
    def annotate(self):
        """
        Returns a list of three element tuples with lineno, commit and line
        """
        if self.commit is None:
            raise NodeError("Unable to get commit for this FileNode")
        pre_load = ["author", "date", "message", "parents"]
        return self.commit.get_file_annotate(self.bytes_path, pre_load=pre_load)

    @LazyProperty
    def is_binary(self):
        """
        Returns True if file has binary content.
        """
        if self.commit:
            return self.commit.is_node_binary(self.bytes_path)
        else:
            raw_bytes = self._content
            return bool(raw_bytes and BIN_BYTE_MARKER in raw_bytes)

    @LazyProperty
    def md5(self):
        """
        Returns md5 of the file node.
        """

        if self.commit:
            return self.commit.node_md5_hash(self.bytes_path)
        else:
            raw_bytes = self._content
            # TODO: this sucks, we're computing md5 on potentially super big stream data...
            return md5(raw_bytes)

    @LazyProperty
    def extension(self):
        """Returns filenode extension"""
        return self.name.split(".")[-1]

    @property
    def is_executable(self):
        """
        Returns ``True`` if file has executable flag turned on.
        """
        return bool(self.mode & stat.S_IXUSR)

    def get_largefile_node(self):
        """
        Try to return a Mercurial FileNode from this node. It does internal
        checks inside largefile store, if that file exist there it will
        create special instance of LargeFileNode which can get content from
        LF store.
        """
        if self.commit:
            return self.commit.get_largefile_node(self.bytes_path)

    def count_lines(self, content: str | bytes, count_empty=False):
        if isinstance(content, str):
            newline_marker = "\n"
        elif isinstance(content, bytes):
            newline_marker = b"\n"
        else:
            raise ValueError("content must be bytes or str got {type(content)} instead")

        if count_empty:
            all_lines = 0
            empty_lines = 0
            for line in content.splitlines(True):
                if line == newline_marker:
                    empty_lines += 1
                all_lines += 1

            return all_lines, all_lines - empty_lines
        else:
            # fast method
            empty_lines = all_lines = content.count(newline_marker)
            if all_lines == 0 and content:
                # one-line without a newline
                empty_lines = all_lines = 1

        return all_lines, empty_lines

    def lines(self, count_empty=False):
        all_lines, empty_lines = 0, 0

        if not self.is_binary:
            content = self.content
            all_lines, empty_lines = self.count_lines(content, count_empty=count_empty)
        return all_lines, empty_lines


class DirNode(Node):
    """
    DirNode stores list of files and directories within this node.
    Nodes may be used standalone but within repository context they
    lazily fetch data within same repository's commit.
    """

    def __init__(self, path, nodes=(), commit=None, default_pre_load=None):
        """
        Only one of ``nodes`` and ``commit`` may be given. Passing both
        would raise ``NodeError`` exception.

        :param path: relative path to the node
        :param nodes: content may be passed to constructor
        :param commit: if given, will use it to lazily fetch content
        """
        if nodes and commit:
            raise NodeError("Cannot use both nodes and commit")
        super().__init__(path, NodeKind.DIR)
        self.commit = commit
        self._nodes = nodes
        self.default_pre_load = default_pre_load or ["is_binary", "size"]

    def __iter__(self):
        yield from self.nodes

    def __eq__(self, other):
        eq = super().__eq__(other)
        if eq is not None:
            return eq
        # check without entering each dir
        self_nodes_paths = list(sorted(n.path for n in self.nodes))
        other_nodes_paths = list(sorted(n.path for n in self.nodes))
        return self_nodes_paths == other_nodes_paths

    def __lt__(self, other):
        lt = super().__lt__(other)
        if lt is not None:
            return lt
        # check without entering each dir
        self_nodes_paths = list(sorted(n.path for n in self.nodes))
        other_nodes_paths = list(sorted(n.path for n in self.nodes))
        return self_nodes_paths < other_nodes_paths

    @LazyProperty
    def content(self):
        raise NodeError(f"{self} represents a dir and has no `content` attribute")

    @LazyProperty
    def nodes(self):
        if self.commit:
            nodes = self.commit.get_nodes(self.bytes_path, pre_load=self.default_pre_load)
        else:
            nodes = self._nodes
        return sorted(nodes)

    @LazyProperty
    def files(self):
        return sorted(node for node in self.nodes if node.is_file())

    @LazyProperty
    def dirs(self):
        return sorted(node for node in self.nodes if node.is_dir())

    @LazyProperty
    def state(self):
        raise NodeError("Cannot access state of DirNode")

    @LazyProperty
    def size(self):
        size = 0
        for root, dirs, files in self.commit.walk(self.bytes_path):
            for f in files:
                size += f.size

        return size

    @LazyProperty
    def last_commit(self):
        if self.commit:
            pre_load = ["author", "date", "message", "parents"]
            return self.commit.get_path_commit(self.bytes_path, pre_load=pre_load)
        raise NodeError("Cannot retrieve last commit of the file without related commit attribute")

    def __repr__(self):
        short_id = getattr(self.commit, "short_id", "")
        return f"<{self.__class__.__name__} path={self.str_path!r}, short_id={short_id}>"


class RootNode(DirNode):
    """
    DirNode being the root node of the repository.
    """

    def __init__(self, nodes=(), commit=None):
        super().__init__(path=b"", nodes=nodes, commit=commit)

    def __repr__(self):
        short_id = getattr(self.commit, "short_id", "")
        return f"<{self.__class__.__name__} path={self.str_path!r}, short_id={short_id}>"


class SubModuleNode(Node):
    """
    represents a SubModule of Git or SubRepo of Mercurial
    """

    is_binary = False
    size = 0

    def __init__(self, name, url=None, commit=None, alias=None):
        self.path: bytes = name
        self.str_path: str = safe_str(self.path)  # we store paths as str
        self.kind = NodeKind.SUBMODULE
        self.alias = alias

        # we have to use EmptyCommit here since this can point to svn/git/hg
        # submodules we cannot get from repository
        self.commit = EmptyCommit(safe_str(commit), alias=alias)
        self.url = safe_str(url) or self._extract_submodule_url()

    def __repr__(self):
        short_id = getattr(self.commit, "short_id", "")
        return f"<{self.__class__.__name__} {self.str_path!r} @ {short_id}>"

    def _extract_submodule_url(self):
        # TODO: find a way to parse gits submodule file and extract the linking URL
        return safe_str(self.path)

    @LazyProperty
    def name(self):
        """
        Returns name of the node so if its path
        then only last part is returned.
        """
        org = self.str_path.rstrip("/").split("/")[-1]
        return f"{org} @ {self.commit.short_id}"


class LargeFileNode(FileNode):
    def __init__(self, path, url=None, commit=None, alias=None, org_path=None):
        self._validate_path(path)  # can throw exception if path is invalid
        self.org_path = org_path  # as stored in VCS as LF pointer

        self.bytes_path = path.rstrip(b"/")  # store for __repr__
        self.str_path = safe_str(self.bytes_path)
        self.path = self.str_path

        self.kind = NodeKind.LARGE_FILE
        self.alias = alias
        self._content = b""

    def _validate_path(self, path: bytes):
        """
        we override check since the LargeFileNode path is system absolute, but we check for bytes only
        """
        self._assert_bytes(path)

    def __repr__(self):
        return f"<{self.__class__.__name__} {self.org_path} -> {self.str_path!r}>"

    @LazyProperty
    def size(self):
        return os.stat(self.path).st_size

    @LazyProperty
    def raw_bytes(self):
        with open(self.path, "rb") as f:
            content = f.read()
        return content

    @LazyProperty
    def name(self):
        """
        Overwrites name to be the org lf path
        """
        return self.org_path

    def stream_bytes(self):
        with open(self.path, "rb") as stream:
            while True:
                data = stream.read(16 * 1024)
                if not data:
                    break
                yield data