DC44131 2023 Talk - Introduction to CodeQL

app.py

#!/usr/bin/env python
# See https://presentations.geekmasher.dev/2023-05-Defcon44131

from flask import Flask, request, render_template
import psycopg2

app = Flask(__name__)
conn = psycopg2.connect("dbname=workshop user=postgres")


def lookup(data):
    cursor = conn.cursor()
    query = f"SELECT * FROM metadata WHERE name='{data}' OR data='{data}'"
    cursor.execute(query)
    return cursor.fetchall()


@app.route("/")
def index():
    search = request.args.get("search")
    results = lookup(search) 
    return render_template(
        "search.html", results=results
    )
    

if __name__ == "__main__":
    app.run()

Q1_FindSource.ql

/**
 * @name SQL Injection - Find Source
 * @description SQL Injection - Find Source
 * @kind problem
 * @id py/workshop
 * @problem.severity error
 * @security-severity 10.0
 * @sub-severity high
 * @precision high
 * @tags security
 */

import python
import semmle.python.Concepts
import semmle.python.ApiGraphs

/*
 * How do we find the source?
 */

from DataFlow::Node request, Attribute attr
where
  request = API::moduleImport("flask").getMember("request").getAValueReachableFromSource() and
  attr.getObject() = request.asExpr()
select attr, "Source"

Q2_FindSink.ql

/**
 * @name SQL Injection - Find Sink
 * @description SQL Injection - Find Sink
 * @kind problem
 * @id py/workshop
 * @problem.severity error
 * @security-severity 10.0
 * @sub-severity high
 * @precision high
 * @tags security
 */

import python
import semmle.python.Concepts
import semmle.python.ApiGraphs

/*
 * What is the sink?
 */

from CallNode call, DataFlow::Node sink
where
  // Find all functions called "execute"
  call.getFunction().(AttrNode).getName() in ["execute"] and
  // The first argument is what we are interested in
  sink.asCfgNode() = call.getArg(0)
select sink, "Sink"

Q3_SQLInjectionQuery.ql

/**
 * @name SQL Injection
 * @description SQL Injection
 * @kind path-problem
 * @id py/workshop
 * @problem.severity error
 * @security-severity 10.0
 * @sub-severity high
 * @precision high
 * @tags security
 */

import python
import semmle.python.dataflow.new.DataFlow
import semmle.python.dataflow.new.TaintTracking
import semmle.python.Concepts
import semmle.python.ApiGraphs
import DataFlow::PathGraph

/*
 * Can we find a path from Source to Sink?
 */

class Sources extends DataFlow::Node {
  Sources() {
    exists(DataFlow::Node request, Attribute attr |
      request = API::moduleImport("flask").getMember("request").getAValueReachableFromSource() and
      attr.getObject() = request.asExpr() and
      this.asExpr() = attr
    )
  }
}

class Sinks extends DataFlow::Node {
  Sinks() {
    exists(CallNode call, DataFlow::Node sink |
      call.getFunction().(AttrNode).getName() in ["execute"] and
      this.asCfgNode() = call.getArg(0)
    )
  }
}

class SqlInjectionConfig extends TaintTracking::Configuration {
  SqlInjectionConfig() { this = "SqlInjectionConfig" }

  override predicate isSource(DataFlow::Node source) { source instanceof Sources }

  override predicate isSink(DataFlow::Node sink) { sink instanceof Sinks }
}

from SqlInjectionConfig config, DataFlow::PathNode source, DataFlow::PathNode sink
where config.hasFlowPath(source, sink)
select sink.getNode(), source, sink, "This SQL query depends on $@.", source.getNode(),
  "a user-provided value"