# pylint:disable=too-many-boolean-expressions
from __future__ import annotations
import os
import logging
import itertools
from collections import defaultdict
from collections.abc import Iterable
import contextlib
from typing import overload, TYPE_CHECKING

import networkx
import pydemumble

from cle.backends.symbol import Symbol
from archinfo.arch_arm import get_real_address_if_arm
import claripy

from angr.knowledge_plugins.cfg.memory_data import MemoryDataSort
from angr.codenode import CodeNode, BlockNode, HookNode, SyscallNode
from angr.serializable import Serializable
from angr.errors import AngrValueError, SimEngineError, SimMemoryError
from angr.procedures import SIM_LIBRARIES
from angr.procedures.definitions import SimSyscallLibrary
from angr.protos import function_pb2
from angr.calling_conventions import DEFAULT_CC, default_cc
from angr.sim_type import SimTypeFunction, parse_defns
from angr.calling_conventions import SimCC
from angr.project import Project
from angr.utils.library import get_cpp_function_name_and_metadata
from .function_parser import FunctionParser

if TYPE_CHECKING:
    from angr.knowledge_plugins.functions.function_manager import FunctionManager

l = logging.getLogger(name=__name__)


class Function(Serializable):
    """
    A representation of a function and various information about it.
    """

    __slots__ = (
        "_addr_to_block_node",
        "_argument_registers",
        "_argument_stack_variables",
        "_block_sizes",
        "_call_sites",
        "_callout_sites",
        "_cyclomatic_complexity",
        "_endpoints",
        "_function_manager",
        "_jumpout_sites",
        "_local_block_addrs",
        "_local_blocks",
        "_local_transition_graph",
        "_name",
        "_project",
        "_ret_sites",
        "_retout_sites",
        "_returning",
        "addr",
        "addr",
        "binary_name",
        "bp_on_stack",
        "calling_convention",
        "from_signature",
        "info",
        "is_alignment",
        "is_default_name",
        "is_plt",
        "is_prototype_guessed",
        "is_simprocedure",
        "is_syscall",
        "normalized",
        "previous_names",
        "prototype",
        "prototype_libname",
        "ran_cca",
        "retaddr_on_stack",
        "sp_delta",
        "startpoint",
        "tags",
        "transition_graph",
    )

    def __init__(
        self,
        function_manager: FunctionManager | None,
        addr: int,
        name=None,
        syscall=None,
        is_simprocedure: bool | None = None,
        binary_name=None,
        is_plt: bool | None = None,
        returning=None,
        alignment=False,
        calling_convention: SimCC | None = None,
        prototype: SimTypeFunction | None = None,
        prototype_libname: str | None = None,
        is_prototype_guessed: bool = True,
    ):
        """
        Function constructor. If the optional parameters are not provided, they will be automatically determined upon
        the creation of a Function object.

        :param addr:            The address of the function.

        The following parameters are optional.

        :param str name:        The name of the function.
        :param bool syscall:    Whether this function is a syscall or not.
        :param bool is_simprocedure:    Whether this function is a SimProcedure or not.
        :param str binary_name: Name of the binary where this function is.
        :param bool is_plt:     If this function is a PLT entry.
        :param bool returning:  If this function returns.
        :param bool alignment:  If this function acts as an alignment filler. Such functions usually only contain nops.
        """
        self.transition_graph = networkx.classes.digraph.DiGraph()
        self._local_transition_graph = None
        self.normalized = False

        # block nodes at whose ends the function returns
        self._ret_sites: set[CodeNode] = set()
        # block nodes at whose ends the function jumps out to another function (jumps outside)
        self._jumpout_sites: set[CodeNode] = set()
        # block nodes at whose ends the function calls out to another non-returning function
        self._callout_sites: set[CodeNode] = set()
        # block nodes that ends the function by returning out to another function (returns outside). This is rare.
        self._retout_sites: set[CodeNode] = set()
        # block nodes (basic block nodes) at whose ends the function terminates
        # in theory, if everything works fine, endpoints == ret_sites | jumpout_sites | callout_sites
        self._endpoints: defaultdict[str, set[CodeNode]] = defaultdict(set)

        self._call_sites = {}
        self.addr = addr
        # startpoint can be None if the corresponding CFGNode is a syscall node
        self.startpoint = None
        self._function_manager = function_manager
        self.is_syscall = None
        self.is_simprocedure = False
        self.is_alignment = alignment

        # These properties are set by VariableManager
        self.bp_on_stack = False
        self.retaddr_on_stack = False
        self.sp_delta = 0
        # Calling convention
        self.calling_convention = calling_convention
        # Function prototype
        self.prototype = prototype
        self.prototype_libname = prototype_libname
        self.is_prototype_guessed = is_prototype_guessed
        # Whether this function returns or not. `None` means it's not determined yet
        self._returning = None

        self._addr_to_block_node = {}  # map addresses to nodes. it's a cache of blocks. if a block is removed from the
        # function, it may not be removed from _addr_to_block_node. if you want to list
        # all blocks of a function, access .blocks.
        self._block_sizes = {}  # map addresses to block sizes
        self._local_blocks = {}  # a dict of all blocks inside the function
        self._local_block_addrs = set()  # a set of addresses of all blocks inside the function

        self.info = {}  # storing special information, like $gp values for MIPS32
        self.tags = ()  # store function tags. can be set manually by performing CodeTagging analysis.

        # Initialize _cyclomatic_complexity to None
        self._cyclomatic_complexity = None

        # TODO: Can we remove the following two members?
        # Register offsets of those arguments passed in registers
        self._argument_registers = []
        # Stack offsets of those arguments passed in stack variables
        self._argument_stack_variables = []

        self._project: Project | None = None  # will be initialized upon the first access to self.project

        self.ran_cca = False  # this is set by CompleteCallingConventions to avoid reprocessing failed functions

        #
        # Initialize unspecified properties
        #

        if syscall is not None:
            self.is_syscall = syscall
        else:
            if self.project is None:
                raise ValueError(
                    "'syscall' must be specified if you do not specify a function manager for this new function."
                )

            # Determine whether this function is a syscall or not
            self.is_syscall = self.project.simos.is_syscall_addr(addr)

        # Determine whether this function is a SimProcedure
        if is_simprocedure is not None:
            self.is_simprocedure = is_simprocedure
        else:
            if self.project is None:
                raise ValueError(
                    "'is_simprocedure' must be specified if you do not specify a function manager for this new "
                    "function."
                )

            if self.is_syscall or self.project.is_hooked(addr):
                self.is_simprocedure = True

        # Determine if this function is a PLT entry
        if is_plt is not None:
            self.is_plt = is_plt
        else:
            if self._function_manager is not None:
                # use the faster cached version
                self.is_plt = self._function_manager.is_plt_cached(addr)
            else:
                # Whether this function is a PLT entry or not is primarily relying on the PLT detection in CLE; it may
                # also be updated (to True) during CFG recovery.
                if self.project is None:
                    raise ValueError(
                        "'is_plt' must be specified if you do not specify a function manager for this new function."
                    )
                self.is_plt = self.project.loader.find_plt_stub_name(addr) is not None

        # Determine the name of this function
        if name is None:
            self._name = self._get_initial_name()
        else:
            self.is_default_name = False
            self._name = name
        self.previous_names = []
        self.from_signature: str | None = None

        # Determine the name the binary where this function is.
        if binary_name is not None:
            self.binary_name = binary_name
        else:
            self.binary_name = self._get_initial_binary_name()

        # Determine returning status for SimProcedures and Syscalls
        if returning is not None:
            self._returning = returning
        else:
            if self.project is None:
                raise ValueError(
                    "'returning' must be specified if you do not specify a function manager for this new function."
                )

            self._returning = self._get_initial_returning()

        self._init_prototype_and_calling_convention()

    @property
    def name(self):
        return self._name

    @name.setter
    def name(self, v):
        self.previous_names.append(self._name)
        self._name = v
        self._function_manager._kb.labels[self.addr] = v

    @property
    def project(self):
        if self._project is None and self._function_manager is not None:
            # try to set it from function manager
            self._project = self._function_manager._kb._project
        return self._project

    @property
    def returning(self):
        return self._returning

    @returning.setter
    def returning(self, v):
        self._returning = v

    @property
    def blocks(self):
        """
        An iterator of all local blocks in the current function.

        :return: angr.lifter.Block instances.
        """

        for block_addr, block in self._local_blocks.items():
            with contextlib.suppress(SimEngineError, SimMemoryError):
                yield self.get_block(
                    block_addr, size=block.size, byte_string=block.bytestr if isinstance(block, BlockNode) else None
                )

    @property
    def cyclomatic_complexity(self):
        """
        The cyclomatic complexity of the function.

        Cyclomatic complexity is a software metric used to indicate the complexity of a program.
        It is a quantitative measure of the number of linearly independent paths through a program's source code.
        It is computed using the formula: M = E - N + 2P, where
        E = the number of edges in the graph,
        N = the number of nodes in the graph,
        P = the number of connected components.

        The cyclomatic complexity value is lazily computed and cached for future use.
        Initially this value is None until it is computed for the first time

        :return: The cyclomatic complexity of the function.
        :rtype: int
        """
        if self._cyclomatic_complexity is None:
            self._cyclomatic_complexity = (
                self.transition_graph.number_of_edges() - self.transition_graph.number_of_nodes() + 2
            )
        return self._cyclomatic_complexity

    @property
    def xrefs(self):
        """
        An iterator of all xrefs of the current function.

        :return: angr.knowledge_plugins.xrefs.xref.XRef instances.
        """
        for block in self.blocks:
            yield from self._function_manager._kb.xrefs.get_xrefs_by_ins_addr_region(
                block.addr, block.addr + block.size
            )

    @property
    def block_addrs(self):
        """
        An iterator of all local block addresses in the current function.

        :return: block addresses.
        """

        return self._local_blocks.keys()

    @property
    def block_addrs_set(self):
        """
        Return a set of block addresses for a better performance of inclusion tests.

        :return: A set of block addresses.
        :rtype: set
        """

        return self._local_block_addrs

    def get_block(self, addr: int, size: int | None = None, byte_string: bytes | None = None):
        """
        Getting a block out of the current function.

        :param int addr:    The address of the block.
        :param int size:    The size of the block. This is optional. If not provided, angr will load
        :param byte_string:
        :return:
        """
        if size is None and addr in self.block_addrs:
            # we know the size
            size = self._block_sizes[addr]

        assert self.project is not None
        block = self.project.factory.block(addr, size=size, byte_string=byte_string)
        if size is None:
            # update block_size dict
            self._block_sizes[addr] = block.size
        return block

    # compatibility
    _get_block = get_block

    def get_block_size(self, addr: int) -> int | None:
        return self._block_sizes.get(addr, None)

    @property
    def nodes(self) -> Iterable[CodeNode]:
        return self.transition_graph.nodes()

    def get_node(self, addr) -> BlockNode | None:
        return self._addr_to_block_node.get(addr, None)

    @property
    def has_unresolved_jumps(self):
        for addr in self.block_addrs:
            if addr in self._function_manager._kb.unresolved_indirect_jumps:
                b = self._function_manager._kb._project.factory.block(addr)
                if b.vex.jumpkind == "Ijk_Boring":
                    return True
        return False

    @property
    def has_unresolved_calls(self):
        for addr in self.block_addrs:
            if addr in self._function_manager._kb.unresolved_indirect_jumps:
                b = self._function_manager._kb._project.factory.block(addr)
                if b.vex.jumpkind == "Ijk_Call":
                    return True
        return False

    @property
    def operations(self):
        """
        All of the operations that are done by this functions.
        """
        return [op for block in self.blocks for op in block.vex.operations]

    @property
    def code_constants(self):
        """
        All of the constants that are used by this functions's code.
        """
        # TODO: remove link register values
        return [const.value for block in self.blocks for const in block.vex.constants]

    @classmethod
    def _get_cmsg(cls):
        return function_pb2.Function()  # pylint:disable=no-member

    def serialize_to_cmessage(self):
        return FunctionParser.serialize(self)

    @classmethod
    def parse_from_cmessage(cls, cmsg, **kwargs):
        """
        :param cmsg:

        :return Function: The function instantiated out of the cmsg data.
        """
        return FunctionParser.parse_from_cmsg(cmsg, **kwargs)

    def string_references(self, minimum_length=2):
        """
        All of the constant string references used by this function.

        :param minimum_length:  The minimum length of strings to find (default is 1)
        :return:                A generator yielding tuples of (address, string) where is address
                                is the location of the string in memory.
        """

        cfg = self._function_manager._kb.cfgs.get_most_accurate()

        for x in self.xrefs:
            try:
                md = cfg.memory_data[x.dst]
            except KeyError:
                continue
            if md.sort not in {MemoryDataSort.String, MemoryDataSort.UnicodeString}:
                continue
            if len(md.content) < minimum_length:
                continue

            yield (md.addr, md.content)

    @property
    def local_runtime_values(self):
        """
        Tries to find all runtime values of this function which do not come from inputs.
        These values are generated by starting from a blank state and reanalyzing the basic blocks once each.
        Function calls are skipped, and back edges are never taken so these values are often unreliable,
        This function is good at finding simple constant addresses which the function will use or calculate.

        :return: a set of constants
        """
        constants = set()

        assert self.project is not None
        if not self.project.loader.main_object.contains_addr(self.addr):
            return constants

        # FIXME the old way was better for architectures like mips, but we need the initial irsb
        # reanalyze function with a new initial state (use persistent registers)
        # initial_state = self._function_manager._cfg.get_any_irsb(self.addr).initial_state
        # fresh_state = self.project.factory.blank_state(mode="fastpath")
        # for reg in initial_state.arch.persistent_regs + ['ip']:
        #     fresh_state.registers.store(reg, initial_state.registers.load(reg))

        # reanalyze function with a new initial state
        fresh_state = self.project.factory.blank_state(mode="fastpath")
        fresh_state.regs.ip = self.addr

        graph_addrs = {x.addr for x in self.graph.nodes() if isinstance(x, BlockNode)}

        # process the nodes in a breadth-first order keeping track of which nodes have already been analyzed
        analyzed = set()
        q = [fresh_state]
        analyzed.add(fresh_state.solver.eval(fresh_state.ip))
        while len(q) > 0:
            state = q.pop()
            # make sure its in this function
            if state.solver.eval(state.ip) not in graph_addrs:
                continue
            # don't trace into simprocedures
            if self.project.is_hooked(state.solver.eval(state.ip)):
                continue
            # don't trace outside of the binary
            if not self.project.loader.main_object.contains_addr(state.solver.eval(state.ip)):
                continue
            # don't trace unreachable blocks
            if state.history.jumpkind in {
                "Ijk_EmWarn",
                "Ijk_NoDecode",
                "Ijk_MapFail",
                "Ijk_NoRedir",
                "Ijk_SigTRAP",
                "Ijk_SigSEGV",
                "Ijk_ClientReq",
            }:
                continue

            curr_ip = state.solver.eval(state.ip)

            # get runtime values from logs of successors
            successors = self.project.factory.successors(state)
            for succ in successors.flat_successors + successors.unsat_successors:
                for a in succ.history.recent_actions:
                    for ao in a.all_objects:
                        if not isinstance(ao.ast, claripy.ast.Base):
                            constants.add(ao.ast)
                        elif not ao.ast.symbolic:
                            constants.add(succ.solver.eval(ao.ast))

                # add successors to the queue to analyze
                if not succ.solver.symbolic(succ.ip):
                    succ_ip = succ.solver.eval(succ.ip)
                    if succ_ip in self and succ_ip not in analyzed:
                        analyzed.add(succ_ip)
                        q.insert(0, succ)

            # force jumps to missing successors
            # (this is a slightly hacky way to force it to explore all the nodes in the function)
            node = self.get_node(curr_ip)
            if node is None:
                # the node does not exist. maybe it's not a block node.
                continue
            missing = {x.addr for x in list(self.graph.successors(node))} - analyzed
            for succ_addr in missing:
                l.info("Forcing jump to missing successor: %#x", succ_addr)
                if succ_addr not in analyzed:
                    all_successors = (
                        successors.unconstrained_successors + successors.flat_successors + successors.unsat_successors
                    )
                    if len(all_successors) > 0:
                        # set the ip of a copied successor to the successor address
                        succ = all_successors[0].copy()
                        succ.ip = succ_addr
                        analyzed.add(succ_addr)
                        q.insert(0, succ)
                    else:
                        l.warning("Could not reach successor: %#x", succ_addr)

        return constants

    @property
    def num_arguments(self):
        return len(self._argument_registers) + len(self._argument_stack_variables)

    def __contains__(self, val):
        if isinstance(val, int):
            return val in self._block_sizes
        return False

    def __str__(self):
        return (
            f"Function {self.name} [{self.addr:#x}]\n"
            f"  Syscall: {self.is_syscall}\n"
            f"  SP difference: {self.sp_delta}\n"
            f"  Has return: {self.has_return}\n"
            f"  Returning: {'Unknown' if self.returning is None else self.returning}\n"
            f"  Alignment: {self.is_alignment}\n"
            f"  Arguments: reg: {self._argument_registers}, stack: {self._argument_stack_variables}\n"
            f"  Blocks: [{', '.join(f'{i:#x}' for i in self.block_addrs)}]\n"
            f"  Cyclomatic Complexity: {self.cyclomatic_complexity}\n"
            f"  Calling convention: {self.calling_convention}"
        )

    def __repr__(self):
        if self.is_syscall:
            return f"<Syscall function {self.name} ({hex(self.addr) if isinstance(self.addr, int) else self.addr})>"
        return f"<Function {self.name} ({hex(self.addr) if isinstance(self.addr, int) else self.addr})>"

    def __setstate__(self, state):
        for k, v in state.items():
            setattr(self, k, v)

    def __getstate__(self):
        # self._local_transition_graph is a cache. don't pickle it
        d = {k: getattr(self, k) for k in self.__slots__}
        d["_local_transition_graph"] = None
        d["_project"] = None
        d["_function_manager"] = None
        return d

    @property
    def endpoints(self):
        return list(itertools.chain(*self._endpoints.values()))

    @property
    def endpoints_with_type(self):
        return self._endpoints

    @property
    def ret_sites(self):
        return list(self._ret_sites)

    @property
    def jumpout_sites(self):
        return list(self._jumpout_sites)

    @property
    def retout_sites(self):
        return list(self._retout_sites)

    @property
    def callout_sites(self):
        return list(self._callout_sites)

    @property
    def size(self):
        return sum(self._block_sizes[addr] for addr in self._local_blocks)

    @property
    def binary(self):
        """
        Get the object this function belongs to.
        :return: The object this function belongs to.
        """
        assert self.project is not None
        return self.project.loader.find_object_containing(self.addr, membership_check=False)

    @property
    def offset(self) -> int:
        """
        :return: the function's binary offset (i.e., non-rebased address)
        """
        return self.addr - self.binary.mapped_base

    @property
    def symbol(self) -> None | Symbol:
        """
        :return: the function's Symbol, if any
        """
        return self.binary.loader.find_symbol(self.addr)

    @property
    def pseudocode(self) -> str | None:
        """
        :return: the function's pseudocode
        """
        if self.project is None:
            l.error("Cannot generate pseudocode because the function is not associated with any angr project.")
            return None
        dec = self.project.analyses.Decompiler(self, cfg=self._function_manager._kb.cfgs.get_most_accurate())
        return dec.codegen.text if dec.codegen else None

    def add_jumpout_site(self, node: CodeNode):
        """
        Add a custom jumpout site.

        :param node:    The address of the basic block that control flow leaves during this transition.
        :return:        None
        """

        node = self._register_node(True, node)
        self._jumpout_sites.add(node)
        self._add_endpoint(node, "transition")

    def add_retout_site(self, node: CodeNode):
        """
        Add a custom retout site.

        Retout (returning to outside of the function) sites are very rare. It mostly occurs during CFG recovery when we
        incorrectly identify the beginning of a function in the first iteration, and then correctly identify that
        function later in the same iteration (function alignments can lead to this bizarre case). We will mark all edges
        going out of the header of that function as a outside edge, because all successors now belong to the
        incorrectly-identified function. This identification error will be fixed in the second iteration of CFG
        recovery. However, we still want to keep track of jumpouts/retouts during the first iteration so other logic in
        CFG recovery still work.

        :param node: The address of the basic block that control flow leaves the current function after a call.
        :return:     None
        """

        node = self._register_node(True, node)
        self._retout_sites.add(node)
        self._add_endpoint(node, "return")

    def _get_initial_name(self):
        """
        Determine the most suitable name of the function.

        :return:    The initial function name.
        :rtype:     string
        """

        name = None
        addr = self.addr

        self.is_default_name = False
        # Try to get a name from existing labels
        if self._function_manager is not None and addr in self._function_manager._kb.labels:
            name = self._function_manager._kb.labels[addr]

        # try to get the name from a hook
        if name is None and self.project is not None:
            project = self.project
            if project.is_hooked(addr):
                hooker = project.hooked_by(addr)
                if hooker is not None:
                    name = hooker.display_name
            elif project.simos.is_syscall_addr(addr):
                syscall_inst = project.simos.syscall_from_addr(addr)
                if syscall_inst is not None:
                    name = syscall_inst.display_name

        # generate an IDA-style sub_X name
        if name is None:
            self.is_default_name = True
            name = f"sub_{addr:x}"

        return name

    def _get_initial_binary_name(self) -> str | None:
        """
        Determine the name of the binary where this function is.

        :return: None
        """

        binary_name = None

        # if this function is a simprocedure but not a syscall, use its library name as
        # its binary name
        # if it is a syscall, fall back to use self.binary.binary which explicitly says cle##kernel
        if self.project and self.is_simprocedure and not self.is_syscall:
            hooker = self.project.hooked_by(self.addr)
            if hooker is not None:
                binary_name = hooker.library_name

        if binary_name is None:
            if self._function_manager is not None:
                # use the faster cached version
                binary_name = self._function_manager.get_binary_name_cached(self.addr)
            else:
                if self.binary is not None and self.binary.binary:
                    binary_name = os.path.basename(self.binary.binary)

        return binary_name

    def _get_initial_returning(self):
        """
        Determine if this function returns or not *if it is hooked by a SimProcedure or a user hook*.

        :return:    True if the hooker returns, False otherwise.
        :rtype:     bool
        """

        hooker = None
        assert self.project is not None
        if self.is_syscall:
            hooker = self.project.simos.syscall_from_addr(self.addr)
        elif self.is_simprocedure:
            hooker = self.project.hooked_by(self.addr)
        if hooker:
            if hasattr(hooker, "DYNAMIC_RET") and hooker.DYNAMIC_RET:
                return True
            return hooker.returns

        # Cannot determine
        return None

    def _init_prototype_and_calling_convention(self) -> None:
        """
        Initialize prototype and calling convention from a SimProcedure, if available.
        """
        hooker = None
        if self.is_syscall and self.project is not None and self.project.simos.is_syscall_addr(self.addr):
            hooker = self.project.simos.syscall_from_addr(self.addr)
        elif self.is_simprocedure and self.project is not None:
            hooker = self.project.hooked_by(self.addr)
        if hooker is None or hooker.guessed_prototype:
            return

        if hooker.prototype:
            self.prototype_libname = hooker.library_name
            self.prototype = hooker.prototype
            self.is_prototype_guessed = False

        cc = hooker.cc
        if cc is None and self.project is not None:
            arch = self.project.arch
            if arch.name in DEFAULT_CC:
                cc_cls = default_cc(
                    arch.name, platform=self.project.simos.name if self.project.simos is not None else None
                )
                if cc_cls is not None:
                    cc = cc_cls(arch)
        self.calling_convention = cc

    def _clear_transition_graph(self):
        self._block_sizes = {}
        self._addr_to_block_node = {}
        self._local_blocks = {}
        self._local_block_addrs = set()
        self.startpoint = None
        self.transition_graph = networkx.classes.digraph.DiGraph()
        self._local_transition_graph = None

        self._ret_sites = set()
        self._jumpout_sites = set()
        self._callout_sites = set()
        self._retout_sites = set()
        self._endpoints = defaultdict(set)
        self._call_sites = {}

    def _confirm_fakeret(self, src, dst):
        if src not in self.transition_graph or dst not in self.transition_graph[src]:
            raise AngrValueError(f"FakeRet edge ({src}, {dst}) is not in transition graph.")

        data = self.transition_graph[src][dst]

        if "type" not in data or data["type"] != "fake_return":
            raise AngrValueError(f"Edge ({src}, {dst}) is not a FakeRet edge")

        # it's confirmed. register the node if needed
        if "outside" not in data or data["outside"] is False:
            dst = self._register_node(True, dst)

        self.transition_graph[src][dst]["confirmed"] = True

    def _transit_to(
        self, from_node: CodeNode, to_node, outside=False, ins_addr=None, stmt_idx=None, is_exception=False
    ):
        """
        Registers an edge between basic blocks in this function's transition graph.
        Arguments are CodeNode objects.

        :param from_node            The address of the basic block that control
                                    flow leaves during this transition.
        :param to_node              The address of the basic block that control
                                    flow enters during this transition.
        :param bool outside:        If this is a transition to another function, e.g. tail call optimization
        :return: None
        """

        if outside:
            from_node = self._register_node(True, from_node)
            if to_node is not None:
                to_node = self._register_node(False, to_node)

            self._jumpout_sites.add(from_node)
        else:
            from_node = self._register_node(True, from_node)
            if to_node is not None:
                to_node = self._register_node(True, to_node)

        type_ = "transition" if not is_exception else "exception"
        if to_node is not None:
            self.transition_graph.add_edge(
                from_node, to_node, type=type_, outside=outside, ins_addr=ins_addr, stmt_idx=stmt_idx
            )

        if outside:
            # this node is an endpoint of the current function
            self._add_endpoint(from_node, type_)

        # clear the cache
        self._local_transition_graph = None

    def _call_to(self, from_node, to_func, ret_node, stmt_idx=None, ins_addr=None, return_to_outside=False):
        """
        Registers an edge between the caller basic block and callee function.

        :param from_addr:   The basic block that control flow leaves during the transition.
