mirror of
https://github.com/github/codeql.git
synced 2025-12-18 01:33:15 +01:00
Port to ApiGraphs and finish the query
This commit is contained in:
@@ -1,192 +0,0 @@
|
||||
/**
|
||||
* @name Python Insecure LDAP Authentication
|
||||
* @description Python LDAP Insecure LDAP Authentication
|
||||
* @kind path-problem
|
||||
* @problem.severity error
|
||||
* @id python/insecure-ldap-auth
|
||||
* @tags experimental
|
||||
* security
|
||||
* external/cwe/cwe-090
|
||||
*/
|
||||
|
||||
// Determine precision above
|
||||
import python
|
||||
import semmle.python.dataflow.new.RemoteFlowSources
|
||||
import semmle.python.dataflow.new.DataFlow
|
||||
import semmle.python.dataflow.new.TaintTracking
|
||||
import semmle.python.dataflow.new.internal.TaintTrackingPublic
|
||||
import DataFlow::PathGraph
|
||||
|
||||
class FalseArg extends ControlFlowNode {
|
||||
FalseArg() { this.getNode().(Expr).(BooleanLiteral) instanceof False }
|
||||
}
|
||||
|
||||
// From luchua-bc's Insecure LDAP authentication in Java (to reduce false positives)
|
||||
string getFullHostRegex() { result = "(?i)ldap://[\\[a-zA-Z0-9].*" }
|
||||
|
||||
string getSchemaRegex() { result = "(?i)ldap(://)?" }
|
||||
|
||||
string getPrivateHostRegex() {
|
||||
result =
|
||||
"(?i)localhost(?:[:/?#].*)?|127\\.0\\.0\\.1(?:[:/?#].*)?|10(?:\\.[0-9]+){3}(?:[:/?#].*)?|172\\.16(?:\\.[0-9]+){2}(?:[:/?#].*)?|192.168(?:\\.[0-9]+){2}(?:[:/?#].*)?|\\[?0:0:0:0:0:0:0:1\\]?(?:[:/?#].*)?|\\[?::1\\]?(?:[:/?#].*)?"
|
||||
}
|
||||
|
||||
// "ldap://somethingon.theinternet.com"
|
||||
class LDAPFullHost extends StrConst {
|
||||
LDAPFullHost() {
|
||||
exists(string s |
|
||||
s = this.getText() and
|
||||
s.regexpMatch(getFullHostRegex()) and
|
||||
not s.substring(7, s.length()).regexpMatch(getPrivateHostRegex()) // No need to check for ldaps, would be SSL by default.
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
class LDAPSchema extends StrConst {
|
||||
LDAPSchema() { this.getText().regexpMatch(getSchemaRegex()) }
|
||||
}
|
||||
|
||||
class LDAPPrivateHost extends StrConst {
|
||||
LDAPPrivateHost() { this.getText().regexpMatch(getPrivateHostRegex()) }
|
||||
}
|
||||
|
||||
predicate concatAndCompareAgainstFullHostRegex(Expr schema, Expr host) {
|
||||
schema instanceof LDAPSchema and
|
||||
not host instanceof LDAPPrivateHost and
|
||||
exists(string full_host |
|
||||
full_host = schema.(StrConst).getText() + host.(StrConst).getText() and
|
||||
full_host.regexpMatch(getFullHostRegex())
|
||||
)
|
||||
}
|
||||
|
||||
// "ldap://" + "somethingon.theinternet.com"
|
||||
class LDAPBothStrings extends BinaryExpr {
|
||||
LDAPBothStrings() { concatAndCompareAgainstFullHostRegex(this.getLeft(), this.getRight()) }
|
||||
}
|
||||
|
||||
// schema + host
|
||||
class LDAPBothVar extends BinaryExpr {
|
||||
LDAPBothVar() {
|
||||
exists(SsaVariable schemaVar, SsaVariable hostVar |
|
||||
this.getLeft() = schemaVar.getVariable().getALoad() and // getAUse is incompatible with Expr
|
||||
this.getRight() = hostVar.getVariable().getALoad() and
|
||||
concatAndCompareAgainstFullHostRegex(schemaVar
|
||||
.getDefinition()
|
||||
.getImmediateDominator()
|
||||
.getNode(), hostVar.getDefinition().getImmediateDominator().getNode())
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// schema + "somethingon.theinternet.com"
|
||||
class LDAPVarString extends BinaryExpr {
|
||||
LDAPVarString() {
|
||||
exists(SsaVariable schemaVar |
|
||||
this.getLeft() = schemaVar.getVariable().getALoad() and
|
||||
concatAndCompareAgainstFullHostRegex(schemaVar
|
||||
.getDefinition()
|
||||
.getImmediateDominator()
|
||||
.getNode(), this.getRight())
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// "ldap://" + host
|
||||
class LDAPStringVar extends BinaryExpr {
|
||||
LDAPStringVar() {
|
||||
exists(SsaVariable hostVar |
|
||||
this.getRight() = hostVar.getVariable().getALoad() and
|
||||
concatAndCompareAgainstFullHostRegex(this.getLeft(),
|
||||
hostVar.getDefinition().getImmediateDominator().getNode())
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
class LDAPInsecureAuthSource extends DataFlow::Node {
|
||||
LDAPInsecureAuthSource() {
|
||||
this instanceof RemoteFlowSource or
|
||||
this.asExpr() instanceof LDAPBothStrings or
|
||||
this.asExpr() instanceof LDAPBothVar or
|
||||
this.asExpr() instanceof LDAPVarString or
|
||||
this.asExpr() instanceof LDAPStringVar
|
||||
}
|
||||
}
|
||||
|
||||
class SafeLDAPOptions extends ControlFlowNode {
|
||||
SafeLDAPOptions() {
|
||||
this = Value::named("ldap.OPT_X_TLS_ALLOW").getAReference() or
|
||||
this = Value::named("ldap.OPT_X_TLS_TRY").getAReference() or
|
||||
this = Value::named("ldap.OPT_X_TLS_DEMAND").getAReference() or
|
||||
this = Value::named("ldap.OPT_X_TLS_HARD").getAReference()
|
||||
}
|
||||
}
|
||||
|
||||
// LDAP3
|
||||
class LDAPInsecureAuthSink extends DataFlow::Node {
|
||||
LDAPInsecureAuthSink() {
|
||||
exists(SsaVariable connVar, CallNode connCall, SsaVariable srvVar, CallNode srvCall |
|
||||
// set connCall as a Call to ldap3.Connection
|
||||
connCall = Value::named("ldap3.Connection").getACall() and
|
||||
// get variable whose definition is a call to ldap3.Connection to correlate ldap3.Server and Connection.start_tls()
|
||||
connVar.getDefinition().getImmediateDominator() = connCall and
|
||||
// get connCall's first argument variable definition
|
||||
srvVar.getAUse() = connCall.getArg(0) and
|
||||
/*
|
||||
* // restrict srvVar definition to a ldap3.Server Call
|
||||
* srvCall = Value::named("ldap3.Server").getACall() and
|
||||
* srvVar.getDefinition().getImmediateDominator() = srvCall
|
||||
* // redundant? ldap3.Connection's first argument *must* be ldap3.Server
|
||||
*/
|
||||
|
||||
// set srvCall as srvVar definition's call
|
||||
srvVar.getDefinition().getImmediateDominator() = srvCall and
|
||||
// set ldap3.Server's 1st argument as sink
|
||||
this.asExpr() = srvCall.getArg(0).getNode() and
|
||||
(
|
||||
// check ldap3.Server call's 3rd argument (positional) is null and there's no use_ssl
|
||||
count(srvCall.getAnArg()) < 3 and
|
||||
count(srvCall.getArgByName("use_ssl")) = 0
|
||||
or
|
||||
// check ldap3.Server call's 3rd argument is False
|
||||
srvCall.getAnArg() instanceof FalseArg
|
||||
or
|
||||
// check ldap3.Server argByName "use_ssl" is False
|
||||
srvCall.getArgByName("use_ssl") instanceof FalseArg
|
||||
) and
|
||||
/*
|
||||
* Avoid flow through any function (Server()) whose variable declaring it (srv) is the first
|
||||
* argument in any function (Connection()) whose variable declaring it also calls .start_tls
|
||||
*/
|
||||
|
||||
/*
|
||||
* host = schema + "somethingon.theinternet.com"
|
||||
* srv = Server(host, port = 1337)
|
||||
* conn = Connection(srv, "dn", "password")
|
||||
* conn.start_tls() !
|
||||
*/
|
||||
|
||||
not connVar
|
||||
.getAUse()
|
||||
.getImmediateDominator()
|
||||
.(CallNode)
|
||||
.getNode()
|
||||
.getFunc()
|
||||
.(Attribute)
|
||||
.getName()
|
||||
.matches("start_tls")
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
class LDAPInsecureAuthConfig extends TaintTracking::Configuration {
|
||||
LDAPInsecureAuthConfig() { this = "LDAPInsecureAuthConfig" }
|
||||
|
||||
override predicate isSource(DataFlow::Node source) { source instanceof LDAPInsecureAuthSource }
|
||||
|
||||
override predicate isSink(DataFlow::Node sink) { sink instanceof LDAPInsecureAuthSink }
|
||||
}
|
||||
|
||||
from LDAPInsecureAuthConfig config, DataFlow::PathNode source, DataFlow::PathNode sink
|
||||
where config.hasFlowPath(source, sink)
|
||||
select sink.getNode(), source, sink, "$@ from $@ is authenticated insecurely.", source.getNode(),
|
||||
"The host", sink.getNode(), "this LDAP query"
|
||||
@@ -1,66 +0,0 @@
|
||||
import ldap
|
||||
from flask import request, Flask
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
schema = "ldap://"
|
||||
remote_host = "127.0.0.1"
|
||||
full_host = "ldap://127.0.0.1"
|
||||
|
||||
|
||||
# GOOD
|
||||
|
||||
# SSL through ldap variable option
|
||||
# host from variable + variable
|
||||
|
||||
@app.route("/one")
|
||||
def one():
|
||||
ldap_connection_1 = ldap.initialize(schema + remote_host)
|
||||
ldap_connection_1.set_option(ldap.OPT_X_TLS_DEMAND, True)
|
||||
ldap_connection_1.simple_bind_s('', '')
|
||||
user = ldap_connection_1.search_s(
|
||||
"dn", ldap.SCOPE_SUBTREE, "search_filter")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
# SSL through ldap start_tls_s
|
||||
# hardcoded host
|
||||
|
||||
@app.route("/two")
|
||||
def two():
|
||||
ldap_connection_2 = ldap.initialize("ldap://127.0.0.1")
|
||||
ldap_connection_2.start_tls_s()
|
||||
ldap_connection_2.simple_bind_s('', '')
|
||||
user = ldap_connection_2.search_s(
|
||||
"dn", ldap.SCOPE_SUBTREE, "search_filter")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
# BAD (not a sink because it's private)
|
||||
|
||||
@app.route("/one_bad")
|
||||
def one_bad():
|
||||
ldap_connection_3 = ldap.initialize(schema + remote_host)
|
||||
ldap_connection_3.set_option(ldap.OPT_X_TLS_DEMAND, False)
|
||||
ldap_connection_3.simple_bind_s('', '')
|
||||
user = ldap_connection_3.search_s(
|
||||
"dn", ldap.SCOPE_SUBTREE, "search_filter")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
@app.route("/one_bad_2")
|
||||
def one_bad_2():
|
||||
ldap_connection_4 = ldap.initialize(schema + remote_host)
|
||||
ldap_connection_4.set_option(ldap.OPT_X_TLS_NEVER)
|
||||
ldap_connection_4.simple_bind_s('', '')
|
||||
user = ldap_connection_4.search_s(
|
||||
"dn", ldap.SCOPE_SUBTREE, "search_filter")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
# if __name__ == "__main__":
|
||||
# app.run(debug=True)
|
||||
@@ -1,66 +0,0 @@
|
||||
import ldap
|
||||
from flask import request, Flask
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
schema = "ldap://"
|
||||
remote_host = "somethingon.theinternet.com"
|
||||
full_host = "ldap://somethingon.theinternet.com"
|
||||
|
||||
|
||||
# GOOD
|
||||
|
||||
# SSL through ldap variable option
|
||||
# host from variable + variable
|
||||
|
||||
@app.route("/one")
|
||||
def one():
|
||||
ldap_connection_5 = ldap.initialize(schema + remote_host)
|
||||
ldap_connection_5.set_option(ldap.OPT_X_TLS_DEMAND, True)
|
||||
ldap_connection_5.simple_bind_s('', '')
|
||||
user = ldap_connection_5.search_s(
|
||||
"dn", ldap.SCOPE_SUBTREE, "search_filter")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
# SSL through ldap start_tls_s
|
||||
# hardcoded host
|
||||
|
||||
@app.route("/two")
|
||||
def two():
|
||||
ldap_connection_6 = ldap.initialize("ldap://somethingon.theinternet.com")
|
||||
ldap_connection_6.start_tls_s()
|
||||
ldap_connection_6.simple_bind_s('', '')
|
||||
user = ldap_connection_6.search_s(
|
||||
"dn", ldap.SCOPE_SUBTREE, "search_filter")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
# BAD
|
||||
|
||||
@app.route("/one_bad")
|
||||
def one_bad():
|
||||
ldap_connection_7 = ldap.initialize(schema + remote_host)
|
||||
ldap_connection_7.set_option(ldap.OPT_X_TLS_DEMAND, False)
|
||||
ldap_connection_7.simple_bind_s('', '')
|
||||
user = ldap_connection_7.search_s(
|
||||
"dn", ldap.SCOPE_SUBTREE, "search_filter")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
@app.route("/one_bad_2")
|
||||
def one_bad_2():
|
||||
ldap_connection_8 = ldap.initialize(schema + remote_host)
|
||||
ldap_connection_8.set_option(ldap.OPT_X_TLS_NEVER)
|
||||
ldap_connection_8.simple_bind_s('', '')
|
||||
user = ldap_connection_8.search_s(
|
||||
"dn", ldap.SCOPE_SUBTREE, "search_filter")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
# if __name__ == "__main__":
|
||||
# app.run(debug=True)
|
||||
@@ -1,105 +0,0 @@
|
||||
from ldap3 import Server, Connection, ALL
|
||||
from flask import request, Flask
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
schema = "ldap://"
|
||||
partial_host = "127.0.0.1"
|
||||
full_host = "ldap://127.0.0.1"
|
||||
|
||||
|
||||
# hardcoded host
|
||||
|
||||
@app.route("/one")
|
||||
def one():
|
||||
srv = Server("ldap://127.0.0.1", port=1337)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# host from variable
|
||||
|
||||
@app.route("/two")
|
||||
def two():
|
||||
srv = Server(full_host, port=1337)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# schema from string + variable
|
||||
|
||||
@app.route("/three")
|
||||
def three():
|
||||
host = "ldap://" + partial_host
|
||||
|
||||
srv = Server(host, port=1337)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# schema from variable + variable
|
||||
|
||||
@app.route("/four")
|
||||
def four():
|
||||
host = schema + partial_host
|
||||
|
||||
srv = Server(host, port=1337)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# schema from string + string
|
||||
|
||||
@app.route("/five")
|
||||
def five():
|
||||
host = "ldap://" + "127.0.0.1"
|
||||
|
||||
srv = Server(host, port=1337)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# schema from variable + hardcoded host
|
||||
|
||||
@app.route("/six")
|
||||
def six():
|
||||
host = schema + "127.0.0.1"
|
||||
|
||||
srv = Server(host, port=1337)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# use_ssl = True (positional argument)
|
||||
# host from string + variable
|
||||
|
||||
@app.route("/four")
|
||||
def four():
|
||||
host = "ldap://" + partial_host
|
||||
|
||||
srv = Server(host, port=1337, True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# use_ssl = True (argument by name)
|
||||
# host from variable + variable
|
||||
|
||||
@app.route("/five")
|
||||
def five():
|
||||
host = schema + partial_host
|
||||
|
||||
srv = Server(host, port=1337, use_ssl=True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
|
||||
|
||||
# if __name__ == "__main__":
|
||||
# app.run(debug=True)
|
||||
@@ -1,146 +0,0 @@
|
||||
from ldap3 import Server, Connection, ALL
|
||||
from flask import request, Flask
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
schema = "ldap://"
|
||||
remote_host = "somethingon.theinternet.com"
|
||||
full_host = "ldap://somethingon.theinternet.com"
|
||||
|
||||
|
||||
# use_ssl = True (positional argument)
|
||||
# hardcoded host
|
||||
|
||||
@app.route("/one")
|
||||
def one():
|
||||
srv = Server("ldap://somethingon.theinternet.com", port=1337, True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# use_ssl = True (argument by name)
|
||||
# host from variable
|
||||
|
||||
@app.route("/two")
|
||||
def two():
|
||||
srv = Server(full_host, port=1337, use_ssl=True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# use_ssl = True (argument by name)
|
||||
# host from RFS
|
||||
|
||||
@app.route("/three")
|
||||
def three():
|
||||
srv = Server(request.args['host'], port=1337, use_ssl=True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# use_ssl = True (positional argument)
|
||||
# host from string + variable
|
||||
|
||||
@app.route("/four")
|
||||
def four():
|
||||
host = "ldap://" + remote_host
|
||||
|
||||
srv = Server(host, port=1337, True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# use_ssl = True (argument by name)
|
||||
# host from variable + variable
|
||||
|
||||
@app.route("/five")
|
||||
def five():
|
||||
host = schema + remote_host
|
||||
|
||||
srv = Server(host, port=1337, use_ssl=True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# use_ssl = True (argument by name)
|
||||
# host from string + RFS
|
||||
|
||||
@app.route("/six")
|
||||
def six():
|
||||
host = "ldap://" + request.args['host']
|
||||
|
||||
srv = Server(host, port=1337, use_ssl=True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# use_ssl = True (positional argument)
|
||||
# host from variable + RFS
|
||||
|
||||
@app.route("/seven")
|
||||
def seven():
|
||||
host = schema + request.args['host']
|
||||
|
||||
srv = Server(host, port=1337, True)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# SSL through special method
|
||||
# host from variable + hardcoded host
|
||||
|
||||
@app.route("/eight")
|
||||
def eight():
|
||||
host = schema + "somethingon.theinternet.com"
|
||||
srv = Server(host, port=1337)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.start_tls() # !
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# No SSL (to test sink)
|
||||
# host from variable + hardcoded host
|
||||
|
||||
@app.route("/nine")
|
||||
def nine():
|
||||
host = schema + "somethingon.theinternet.com"
|
||||
srv = Server(host, port=1337, False)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# No SSL (to test sink)
|
||||
# host from variable + variable
|
||||
|
||||
@app.route("/ten")
|
||||
def ten():
|
||||
host = schema + remote_host
|
||||
srv = Server(host, port=1337, use_ssl=False)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# No SSL (to test sink)
|
||||
# host from variable + RFS
|
||||
|
||||
@app.route("/eleven")
|
||||
def eleven():
|
||||
host = schema + request.args['host']
|
||||
srv = Server(host, port=1337)
|
||||
conn = Connection(srv, "dn", "password")
|
||||
conn.search("dn", "search_filter")
|
||||
return conn.response
|
||||
|
||||
|
||||
# if __name__ == "__main__":
|
||||
# app.run(debug=True)
|
||||
@@ -0,0 +1,20 @@
|
||||
/**
|
||||
* @name Python Insecure LDAP Authentication
|
||||
* @description Python LDAP Insecure LDAP Authentication
|
||||
* @kind path-problem
|
||||
* @problem.severity error
|
||||
* @id python/insecure-ldap-auth
|
||||
* @tags experimental
|
||||
* security
|
||||
* external/cwe/cwe-522
|
||||
*/
|
||||
|
||||
// determine precision above
|
||||
import python
|
||||
import DataFlow::PathGraph
|
||||
import experimental.semmle.python.security.LDAPInsecureAuth
|
||||
|
||||
from LDAPInsecureAuthConfig config, DataFlow::PathNode source, DataFlow::PathNode sink
|
||||
where config.hasFlowPath(source, sink)
|
||||
select sink.getNode(), source, sink, "$@ is authenticated insecurely.", sink.getNode(),
|
||||
"This LDAP host"
|
||||
@@ -156,10 +156,20 @@ module LDAPBind {
|
||||
* extend `LDAPBind` instead.
|
||||
*/
|
||||
abstract class Range extends DataFlow::Node {
|
||||
/**
|
||||
* Gets the argument containing the binding host.
|
||||
*/
|
||||
abstract DataFlow::Node getHost();
|
||||
|
||||
/**
|
||||
* Gets the argument containing the binding expression.
|
||||
*/
|
||||
abstract DataFlow::Node getPassword();
|
||||
|
||||
/**
|
||||
* Checks if the binding process use SSL.
|
||||
*/
|
||||
abstract predicate useSSL();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -174,5 +184,18 @@ class LDAPBind extends DataFlow::Node {
|
||||
|
||||
LDAPBind() { this = range }
|
||||
|
||||
/**
|
||||
* Gets the argument containing the binding host.
|
||||
*/
|
||||
DataFlow::Node getHost() { result = range.getHost() }
|
||||
|
||||
/**
|
||||
* Gets the argument containing the binding expression.
|
||||
*/
|
||||
DataFlow::Node getPassword() { result = range.getPassword() }
|
||||
|
||||
/**
|
||||
* Checks if the binding process use SSL.
|
||||
*/
|
||||
predicate useSSL() { range.useSSL() }
|
||||
}
|
||||
|
||||
@@ -99,6 +99,42 @@ private module LDAP {
|
||||
override DataFlow::Node getPassword() {
|
||||
result in [this.getArg(1), this.getArgByName("cred")]
|
||||
}
|
||||
|
||||
override DataFlow::Node getHost() {
|
||||
exists(DataFlow::CallCfgNode initialize |
|
||||
this.getFunction().(DataFlow::AttrRead).getObject().getALocalSource() = initialize and
|
||||
initialize = ldapInitialize().getACall() and
|
||||
result = initialize.getArg(0)
|
||||
)
|
||||
}
|
||||
|
||||
override predicate useSSL() {
|
||||
// use initialize to correlate `this` and so avoid FP in several instances
|
||||
exists(DataFlow::CallCfgNode initialize |
|
||||
this.getFunction().(DataFlow::AttrRead).getObject().getALocalSource() = initialize and
|
||||
initialize = ldapInitialize().getACall() and
|
||||
(
|
||||
// ldap_connection.start_tls_s()
|
||||
exists(DataFlow::AttrRead startTLS |
|
||||
startTLS.getObject().getALocalSource() = initialize and
|
||||
startTLS.getAttributeName().matches("%start_tls%")
|
||||
)
|
||||
or
|
||||
// ldap_connection.set_option(ldap.OPT_X_TLS_%s, True)
|
||||
// ldap_connection.set_option(ldap.OPT_X_TLS_%s)
|
||||
exists(DataFlow::CallCfgNode setOption |
|
||||
setOption.getFunction().(DataFlow::AttrRead).getObject().getALocalSource() =
|
||||
initialize and
|
||||
setOption.getFunction().(DataFlow::AttrRead).getAttributeName() = "set_option" and
|
||||
setOption.getArg(0) =
|
||||
ldap().getMember("OPT_X_TLS_" + ["ALLOW", "TRY", "DEMAND", "HARD"]).getAUse() and
|
||||
not DataFlow::exprNode(any(False falseExpr))
|
||||
.(DataFlow::LocalSourceNode)
|
||||
.flowsTo(setOption.getArg(1))
|
||||
)
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -166,6 +202,24 @@ private module LDAP {
|
||||
override DataFlow::Node getPassword() {
|
||||
result in [this.getArg(2), this.getArgByName("password")]
|
||||
}
|
||||
|
||||
override DataFlow::Node getHost() {
|
||||
exists(DataFlow::CallCfgNode serverCall |
|
||||
serverCall = ldap3Server().getACall() and
|
||||
this.getArg(0).getALocalSource() = serverCall and
|
||||
result = serverCall.getArg(0)
|
||||
)
|
||||
}
|
||||
|
||||
override predicate useSSL() {
|
||||
exists(DataFlow::CallCfgNode serverCall |
|
||||
serverCall = ldap3Server().getACall() and
|
||||
this.getArg(0).getALocalSource() = serverCall and
|
||||
DataFlow::exprNode(any(True trueExpr))
|
||||
.(DataFlow::LocalSourceNode)
|
||||
.flowsTo([serverCall.getArg(2), serverCall.getArgByName("use_ssl")])
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
/**
|
||||
* Provides a taint-tracking configuration for detecting LDAP injection vulnerabilities
|
||||
*/
|
||||
|
||||
import python
|
||||
import semmle.python.dataflow.new.DataFlow
|
||||
import semmle.python.dataflow.new.TaintTracking
|
||||
import semmle.python.dataflow.new.RemoteFlowSources
|
||||
import experimental.semmle.python.Concepts
|
||||
|
||||
string getFullHostRegex() { result = "(?i)ldap://[\\[a-zA-Z0-9].*" }
|
||||
|
||||
string getSchemaRegex() { result = "(?i)ldap(://)?" }
|
||||
|
||||
string getPrivateHostRegex() {
|
||||
result =
|
||||
"(?i)localhost(?:[:/?#].*)?|127\\.0\\.0\\.1(?:[:/?#].*)?|10(?:\\.[0-9]+){3}(?:[:/?#].*)?|172\\.16(?:\\.[0-9]+){2}(?:[:/?#].*)?|192.168(?:\\.[0-9]+){2}(?:[:/?#].*)?|\\[?0:0:0:0:0:0:0:1\\]?(?:[:/?#].*)?|\\[?::1\\]?(?:[:/?#].*)?"
|
||||
}
|
||||
|
||||
// "ldap://somethingon.theinternet.com"
|
||||
class LDAPFullHost extends StrConst {
|
||||
LDAPFullHost() {
|
||||
exists(string s |
|
||||
s = this.getText() and
|
||||
s.regexpMatch(getFullHostRegex()) and
|
||||
not s.substring(7, s.length()).regexpMatch(getPrivateHostRegex()) // No need to check for ldaps, it would be SSL by default.
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
class LDAPSchema extends StrConst {
|
||||
LDAPSchema() { this.getText().regexpMatch(getSchemaRegex()) }
|
||||
}
|
||||
|
||||
class LDAPPrivateHost extends StrConst {
|
||||
LDAPPrivateHost() { this.getText().regexpMatch(getPrivateHostRegex()) }
|
||||
}
|
||||
|
||||
predicate concatAndCompareAgainstFullHostRegex(Expr schema, Expr host) {
|
||||
schema instanceof LDAPSchema and
|
||||
not host instanceof LDAPPrivateHost and
|
||||
exists(string full_host |
|
||||
full_host = schema.(StrConst).getText() + host.(StrConst).getText() and
|
||||
full_host.regexpMatch(getFullHostRegex())
|
||||
)
|
||||
}
|
||||
|
||||
// "ldap://" + "somethingon.theinternet.com"
|
||||
class LDAPBothStrings extends BinaryExpr {
|
||||
LDAPBothStrings() { concatAndCompareAgainstFullHostRegex(this.getLeft(), this.getRight()) }
|
||||
}
|
||||
|
||||
// schema + host
|
||||
class LDAPBothVar extends BinaryExpr {
|
||||
LDAPBothVar() {
|
||||
exists(SsaVariable schemaVar, SsaVariable hostVar |
|
||||
this.getLeft() = schemaVar.getVariable().getALoad() and // getAUse is incompatible with Expr
|
||||
this.getRight() = hostVar.getVariable().getALoad() and
|
||||
concatAndCompareAgainstFullHostRegex(schemaVar
|
||||
.getDefinition()
|
||||
.getImmediateDominator()
|
||||
.getNode(), hostVar.getDefinition().getImmediateDominator().getNode())
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// schema + "somethingon.theinternet.com"
|
||||
class LDAPVarString extends BinaryExpr {
|
||||
LDAPVarString() {
|
||||
exists(SsaVariable schemaVar |
|
||||
this.getLeft() = schemaVar.getVariable().getALoad() and
|
||||
concatAndCompareAgainstFullHostRegex(schemaVar
|
||||
.getDefinition()
|
||||
.getImmediateDominator()
|
||||
.getNode(), this.getRight())
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// "ldap://" + host
|
||||
class LDAPStringVar extends BinaryExpr {
|
||||
LDAPStringVar() {
|
||||
exists(SsaVariable hostVar |
|
||||
this.getRight() = hostVar.getVariable().getALoad() and
|
||||
concatAndCompareAgainstFullHostRegex(this.getLeft(),
|
||||
hostVar.getDefinition().getImmediateDominator().getNode())
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A taint-tracking configuration for detecting LDAP injections.
|
||||
*/
|
||||
class LDAPInsecureAuthConfig extends TaintTracking::Configuration {
|
||||
LDAPInsecureAuthConfig() { this = "LDAPInsecureAuthConfig" }
|
||||
|
||||
override predicate isSource(DataFlow::Node source) {
|
||||
source instanceof RemoteFlowSource or
|
||||
source.asExpr() instanceof LDAPBothStrings or
|
||||
source.asExpr() instanceof LDAPBothVar or
|
||||
source.asExpr() instanceof LDAPVarString or
|
||||
source.asExpr() instanceof LDAPStringVar
|
||||
}
|
||||
|
||||
override predicate isSink(DataFlow::Node sink) {
|
||||
exists(LDAPBind ldapBind | not ldapBind.useSSL() and sink = ldapBind.getHost())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user