Issue with Calculating Fan-In and Fan-Out for Functions in Large Python Projects ( Django) Using Joern

[angiolye24]

I am encountering an issue with accurately calculating the fan-in and fan-out of specific functions within a large Python project, such as Django. While attempting to analyze the call relationships, I find that Joern struggles to resolve the correct context for certain function calls, especially when dealing with methods that share common names across various classes and modules. Example query:

import java.io.PrintWriter
import java.io.File

import scala.io.Source

object IndirectFanAnalysis {
 

  def readFunctionNamesFromFile(fileName: String): Set[String] = {

    try {

      Source.fromFile(fileName).getLines().filter(_.nonEmpty).toSet

    } catch {

      case e: Exception =>

        println(s"Errore nella lettura del file: ${e.getMessage}")

        Set.empty[String] // Restituisce un Set vuoto in caso di errore

    }

  }

  // Function to find all indirect fan-in with level indication, excluding level 1 and operators
  def findIndirectFanInWithLevels(functionName: String): Map[Int, List[String]] = {
    var currentLevel = 1
    var seenFunctions = Set[String]()
    var functionsAtLevel = cpg.method.fullNameExact(functionName).caller.fullName.l
      .filterNot(fn => fn.startsWith("<operator>") || fn.startsWith("__builtin") ||
  fn.contains("<metaClassAdapter>") ||
  fn.contains("<fakeNew>") || fn.contains("<returnValue>"))
    var result = Map[Int, List[String]]()

    while (functionsAtLevel.nonEmpty) {
      val uniqueFunctions = functionsAtLevel.filterNot(seenFunctions)
      seenFunctions ++= uniqueFunctions
      if (currentLevel > 1) { // Exclude level 1
        result += (currentLevel -> uniqueFunctions.distinct)
      }
      functionsAtLevel = uniqueFunctions.flatMap(fn => cpg.method.fullNameExact(fn).caller.fullName.l)
        .filterNot(fn => fn.startsWith("<operator>") || fn.startsWith("__builtin") ||
  fn.contains("<metaClassAdapter>") ||
  fn.contains("<fakeNew>") || fn.contains("<returnValue>"))
      currentLevel += 1
    }
    result
  }

  // Function to find all indirect fan-out with level indication, excluding level 1 and operators
  def findIndirectFanOutWithLevels(functionName: String): Map[Int, List[String]] = {
    var currentLevel = 1
    var seenFunctions = Set[String]()
    var functionsAtLevel = cpg.method.fullNameExact(functionName).call.callee.fullName.l
      .filterNot(fn => fn.startsWith("<operator>") || fn.startsWith("__builtin") ||
  fn.contains("<metaClassAdapter>") ||
  fn.contains("<fakeNew>") || fn.contains("<returnValue>"))
    var result = Map[Int, List[String]]()

    while (functionsAtLevel.nonEmpty) {
      val uniqueFunctions = functionsAtLevel.filterNot(seenFunctions)
      seenFunctions ++= uniqueFunctions
      if (currentLevel > 1) { // Exclude level 1
        result += (currentLevel -> uniqueFunctions.distinct)
      }
      functionsAtLevel = uniqueFunctions.flatMap(fn => cpg.method.fullNameExact(fn).call.callee.fullName.l)
        .filterNot(fn => fn.startsWith("<operator>") || fn.startsWith("__builtin") ||
  fn.contains("<metaClassAdapter>") ||
  fn.contains("<fakeNew>") || fn.contains("<returnValue>"))
      currentLevel += 1
    }
    result
  }

  // Execute the query and save results to 'esempio.txt' with progress indication
  def executeAnalysis(): Unit = {



    val functionFilterFile = "filter_functions.txt"

    val includedFunctionNames = readFunctionNamesFromFile(functionFilterFile)


    // Initialize a PrintWriter to write to 'esempio.txt'
    val pw = new PrintWriter(new File("esempio.txt"))
    try {
      // Get the list of function names to process
      val functionNames = cpg.method.fullName.l
        .filterNot(functionName => functionName.startsWith("<operator>") || functionName.startsWith("__builtin") ||
  functionName.contains("<metaClassAdapter>") ||
  functionName.contains("<fakeNew>") || !includedFunctionNames.contains(functionName) || functionName.contains("<returnValue>"))
        .distinct

      val totalFunctions = functionNames.size
      var currentFunctionIndex = 0

      // Process each function and display progress
      functionNames.foreach { functionName =>
        currentFunctionIndex += 1
        println(s"Processing function $currentFunctionIndex of $totalFunctions: $functionName")

        val directFanIn = cpg.method.fullNameExact(functionName).caller.fullName
          .filterNot(f => f.startsWith("<operator>") || f.startsWith("__builtin") ||
  f.contains("<metaClassAdapter>") ||
  f.contains("<fakeNew>") || f.contains("<returnValue>"))
          .distinct
          .l

        val indirectFanInLevels = findIndirectFanInWithLevels(functionName)

        val directFanOut = cpg.method.fullNameExact(functionName).call.callee.fullName
          .filterNot(f => f.startsWith("<operator>") || f.startsWith("__builtin") ||
  f.contains("<metaClassAdapter>") ||
  f.contains("<fakeNew>") || f.contains("<returnValue>"))
          .distinct
          .l

        val indirectFanOutLevels = findIndirectFanOutWithLevels(functionName)

        val fanInOutput = indirectFanInLevels.toSeq.sortBy(_._1).map {
          case (depth, functions) => s"  Indirect Fan-in Level $depth: ${functions.mkString(", ")}"
        }.mkString("\n")

        val fanOutOutput = indirectFanOutLevels.toSeq.sortBy(_._1).map {
          case (depth, functions) => s"  Indirect Fan-out Level $depth: ${functions.mkString(", ")}"
        }.mkString("\n")

        val output = s"""Function: $functionName
  Direct Fan-in Functions: ${directFanIn.mkString(", ")}
$fanInOutput
  Direct Fan-out Functions: ${directFanOut.mkString(", ")}
$fanOutOutput\n"""

        pw.println(output) // Write to 'esempio.txt'
      }
    } finally {
      pw.close() // Close the PrintWriter to ensure data is flushed and resources are released
    }
  }
}

// Run the analysis
IndirectFanAnalysis.executeAnalysis()