Merge pull request github#5284 from yoff/python-port-insecure-protocol

Python: port py/insecure-protocol

Author: RasmusWL

python/change-notes/2021-03-15-port-insecure-protocol.md

@@ -0,0 +1,2 @@
+lgtm,codescanning
+* Ported use of insecure SSL/TLS version (`py/insecure-protocol`) query to use new data-flow library. This might result in different results, but overall a more robust and accurate analysis.

python/ql/src/Security/CWE-327/FluentApiModel.qll

@@ -0,0 +1,103 @@
+private import python
+private import semmle.python.dataflow.new.DataFlow
+import TlsLibraryModel
+
+/**
+ * Configuration to determine the state of a context being used to create
+ * a connection. There is one configuration for each pair of `TlsLibrary` and `ProtocolVersion`,
+ * such that a single configuration only tracks contexts where a specific `ProtocolVersion` is allowed.
+ *
+ * The state is in terms of whether a specific protocol is allowed. This is
+ * either true or false when the context is created and can then be modified
+ * later by either restricting or unrestricting the protocol (see the predicates
+ * `isRestriction` and `isUnrestriction`).
+ *
+ * Since we are interested in the final state, we want the flow to start from
+ * the last unrestriction, so we disallow flow into unrestrictions. We also
+ * model the creation as an unrestriction of everything it allows, to account
+ * for the common case where the creation plays the role of "last unrestriction".
+ *
+ * Since we really want "the last unrestriction, not nullified by a restriction",
+ * we also disallow flow into restrictions.
+ */
+class InsecureContextConfiguration extends DataFlow::Configuration {
+  TlsLibrary library;
+  ProtocolVersion tracked_version;
+
+  InsecureContextConfiguration() {
+    this = library + "Allows" + tracked_version and
+    tracked_version.isInsecure()
+  }
+
+  ProtocolVersion getTrackedVersion() { result = tracked_version }
+
+  override predicate isSource(DataFlow::Node source) { this.isUnrestriction(source) }
+
+  override predicate isSink(DataFlow::Node sink) {
+    sink = library.connection_creation().getContext()
+  }
+
+  override predicate isBarrierIn(DataFlow::Node node) {
+    this.isRestriction(node)
+    or
+    this.isUnrestriction(node)
+  }
+
+  private predicate isRestriction(DataFlow::Node node) {
+    exists(ProtocolRestriction r |
+      r = library.protocol_restriction() and
+      r.getRestriction() = tracked_version
+    |
+      node = r.getContext()
+    )
+  }
+
+  private predicate isUnrestriction(DataFlow::Node node) {
+    exists(ProtocolUnrestriction pu |
+      pu = library.protocol_unrestriction() and
+      pu.getUnrestriction() = tracked_version
+    |
+      node = pu.getContext()
+    )
+  }
+}
+
+/**
+ * Holds if `conectionCreation` marks the creation of a connetion based on the contex
+ * found at `contextOrigin` and allowing `insecure_version`.
+ *
+ * `specific` is true iff the context is configured for a specific protocol version (`ssl.PROTOCOL_TLSv1_2`) rather
+ * than for a family of protocols (`ssl.PROTOCOL_TLS`).
+ */
+predicate unsafe_connection_creation_with_context(
+  DataFlow::Node connectionCreation, ProtocolVersion insecure_version, DataFlow::Node contextOrigin,
+  boolean specific
+) {
+  // Connection created from a context allowing `insecure_version`.
+  exists(InsecureContextConfiguration c | c.hasFlow(contextOrigin, connectionCreation) |
+    insecure_version = c.getTrackedVersion() and
+    specific = false
+  )
+  or
+  // Connection created from a context specifying `insecure_version`.
+  exists(TlsLibrary l |
+    connectionCreation = l.insecure_connection_creation(insecure_version) and
+    contextOrigin = connectionCreation and
+    specific = true
+  )
+}
+
+/**
+ * Holds if `conectionCreation` marks the creation of a connetion witout reference to a context
+ * and allowing `insecure_version`.
+ */
+predicate unsafe_connection_creation_without_context(
+  DataFlow::CallCfgNode connectionCreation, string insecure_version
+) {
+  exists(TlsLibrary l | connectionCreation = l.insecure_connection_creation(insecure_version))
+}
+
+/** Holds if `contextCreation` is creating a context tied to a specific insecure version. */
+predicate unsafe_context_creation(DataFlow::CallCfgNode contextCreation, string insecure_version) {
+  exists(TlsLibrary l | contextCreation = l.insecure_context_creation(insecure_version))
+}

python/ql/src/Security/CWE-327/InsecureProtocol.qhelp

@@ -13,8 +13,8 @@
 
           <p>
             Ensure that a modern, strong protocol is used. All versions of SSL,
-            and TLS 1.0 are known to be vulnerable to attacks. Using TLS 1.1 or
-            above is strongly recommended.
+            and TLS versions 1.0 and 1.1 are known to be vulnerable to attacks.
+            Using TLS 1.2 or above is strongly recommended.
           </p>
 
      </recommendation>
@@ -30,20 +30,35 @@
 
           <p>
             All cases should be updated to use a secure protocol, such as
-            <code>PROTOCOL_TLSv1_1</code>.
+            <code>PROTOCOL_TLSv1_2</code>.
           </p>
           <p>
             Note that <code>ssl.wrap_socket</code> has been deprecated in
-            Python 3.7. A preferred alternative is to use
-            <code>ssl.SSLContext</code>, which is supported in Python 2.7.9 and
-            3.2 and later versions.
+            Python 3.7. The recommended alternatives are:
           </p>
+          <ul>
+            <li><code>ssl.SSLContext</code> - supported in Python 2.7.9,
+                  3.2, and later versions</li>
+            <li><code>ssl.create_default_context</code> - a convenience function,
+                  supported in Python 3.4 and later versions.</li>
+          </ul>
+          <p>
+            Even when you use these alternatives, you should
+            ensure that a safe protocol is used. The following code illustrates
+            how to use flags (available since Python 3.2) or the `minimum_version`
+            field (favored since Python 3.7) to restrict the protocols accepted when
+            creating a connection.
+          </p>
+
+          <sample src="examples/secure_default_protocol.py" />
      </example>
 
      <references>
        <li>Wikipedia: <a href="https://en.wikipedia.org/wiki/Transport_Layer_Security"> Transport Layer Security</a>.</li>
        <li>Python 3 documentation: <a href="https://docs.python.org/3/library/ssl.html#ssl.SSLContext"> class ssl.SSLContext</a>.</li>
        <li>Python 3 documentation: <a href="https://docs.python.org/3/library/ssl.html#ssl.wrap_socket"> ssl.wrap_socket</a>.</li>
+       <li>Python 3 documentation: <a href="https://docs.python.org/3/library/ssl.html#functions-constants-and-exceptions"> notes on context creation</a>.</li>
+       <li>Python 3 documentation: <a href="https://docs.python.org/3/library/ssl.html#ssl-security"> notes on security considerations</a>.</li>
        <li>pyOpenSSL documentation: <a href="https://pyopenssl.org/en/stable/api/ssl.html"> An interface to the SSL-specific parts of OpenSSL</a>.</li>
      </references>
 

python/ql/src/Security/CWE-327/InsecureProtocol.ql

@@ -10,86 +10,77 @@
  */
 
 import python
+import semmle.python.dataflow.new.DataFlow
+import FluentApiModel
 
-private ModuleValue the_ssl_module() { result = Module::named("ssl") }
-
-FunctionValue ssl_wrap_socket() { result = the_ssl_module().attr("wrap_socket") }
-
-ClassValue ssl_Context_class() { result = the_ssl_module().attr("SSLContext") }
+// Helper for pretty printer `configName`.
+// This is a consequence of missing pretty priting.
+// We do not want to evaluate our bespoke pretty printer
+// for all `DataFlow::Node`s so we define a sub class of interesting ones.
+class ProtocolConfiguration extends DataFlow::Node {
+  ProtocolConfiguration() {
+    unsafe_connection_creation_with_context(_, _, this, _)
+    or
+    unsafe_connection_creation_without_context(this, _)
+    or
+    unsafe_context_creation(this, _)
+  }
 
-private ModuleValue the_pyOpenSSL_module() { result = Value::named("pyOpenSSL.SSL") }
+  AstNode getNode() { result = this.asCfgNode().(CallNode).getFunction().getNode() }
+}
 
-ClassValue the_pyOpenSSL_Context_class() { result = Value::named("pyOpenSSL.SSL.Context") }
+// Helper for pretty printer `callName`.
+// This is a consequence of missing pretty priting.
+// We do not want to evaluate our bespoke pretty printer
+// for all `AstNode`s so we define a sub class of interesting ones.
+//
+// Note that AstNode is abstract and AstNode_ is a library class, so
+// we have to extend @py_ast_node.
+class Nameable extends @py_ast_node {
+  Nameable() {
+    this = any(ProtocolConfiguration pc).getNode()
+    or
+    exists(Nameable attr | this = attr.(Attribute).getObject())
+  }
 
-string insecure_version_name() {
-  // For `pyOpenSSL.SSL`
-  result = "SSLv2_METHOD" or
-  result = "SSLv23_METHOD" or
-  result = "SSLv3_METHOD" or
-  result = "TLSv1_METHOD" or
-  // For the `ssl` module
-  result = "PROTOCOL_SSLv2" or
-  result = "PROTOCOL_SSLv3" or
-  result = "PROTOCOL_SSLv23" or
-  result = "PROTOCOL_TLS" or
-  result = "PROTOCOL_TLSv1"
+  string toString() { result = "AstNode" }
 }
 
-/*
- * A syntactic check for cases where points-to analysis cannot infer the presence of
- * a protocol constant, e.g. if it has been removed in later versions of the `ssl`
- * library.
- */
-
-bindingset[named_argument]
-predicate probable_insecure_ssl_constant(
-  CallNode call, string insecure_version, string named_argument
-) {
-  exists(ControlFlowNode arg |
-    arg = call.getArgByName(named_argument) or
-    arg = call.getArg(0)
-  |
-    arg.(AttrNode).getObject(insecure_version).pointsTo(the_ssl_module())
-    or
-    arg.(NameNode).getId() = insecure_version and
-    exists(Import imp |
-      imp.getAnImportedModuleName() = "ssl" and
-      imp.getAName().getAsname().(Name).getId() = insecure_version
-    )
-  )
+string callName(Nameable call) {
+  result = call.(Name).getId()
+  or
+  exists(Attribute a | a = call | result = callName(a.getObject()) + "." + a.getName())
 }
 
-predicate unsafe_ssl_wrap_socket_call(
-  CallNode call, string method_name, string insecure_version, string named_argument
-) {
-  (
-    call = ssl_wrap_socket().getACall() and
-    method_name = "deprecated method ssl.wrap_socket" and
-    named_argument = "ssl_version"
-    or
-    call = ssl_Context_class().getACall() and
-    named_argument = "protocol" and
-    method_name = "ssl.SSLContext"
-  ) and
-  insecure_version = insecure_version_name() and
-  (
-    call.getArgByName(named_argument).pointsTo(the_ssl_module().attr(insecure_version))
-    or
-    probable_insecure_ssl_constant(call, insecure_version, named_argument)
-  )
+string configName(ProtocolConfiguration protocolConfiguration) {
+  result =
+    "call to " + callName(protocolConfiguration.asCfgNode().(CallNode).getFunction().getNode())
+  or
+  not protocolConfiguration.asCfgNode() instanceof CallNode and
+  not protocolConfiguration instanceof ContextCreation and
+  result = "context modification"
 }
 
-predicate unsafe_pyOpenSSL_Context_call(CallNode call, string insecure_version) {
-  call = the_pyOpenSSL_Context_class().getACall() and
-  insecure_version = insecure_version_name() and
-  call.getArg(0).pointsTo(the_pyOpenSSL_module().attr(insecure_version))
+string verb(boolean specific) {
+  specific = true and result = "specified"
+  or
+  specific = false and result = "allowed"
 }
 
-from CallNode call, string method_name, string insecure_version
+from
+  DataFlow::Node connectionCreation, string insecure_version, DataFlow::Node protocolConfiguration,
+  boolean specific
 where
-  unsafe_ssl_wrap_socket_call(call, method_name, insecure_version, _)
+  unsafe_connection_creation_with_context(connectionCreation, insecure_version,
+    protocolConfiguration, specific)
+  or
+  unsafe_connection_creation_without_context(connectionCreation, insecure_version) and
+  protocolConfiguration = connectionCreation and
+  specific = true
   or
-  unsafe_pyOpenSSL_Context_call(call, insecure_version) and method_name = "pyOpenSSL.SSL.Context"
-select call,
-  "Insecure SSL/TLS protocol version " + insecure_version + " specified in call to " + method_name +
-    "."
+  unsafe_context_creation(protocolConfiguration, insecure_version) and
+  connectionCreation = protocolConfiguration and
+  specific = true
+select connectionCreation,
+  "Insecure SSL/TLS protocol version " + insecure_version + " " + verb(specific) + " by $@ ",
+  protocolConfiguration, configName(protocolConfiguration)

python/ql/src/Security/CWE-327/PyOpenSSL.qll

@@ -0,0 +1,83 @@
+/**
+ * Provides modeling of SSL/TLS functionality of the `OpenSSL` module from the `pyOpenSSL` PyPI package.
+ * See https://www.pyopenssl.org/en/stable/
+ */
+
+private import python
+private import semmle.python.ApiGraphs
+import TlsLibraryModel
+
+class PyOpenSSLContextCreation extends ContextCreation, DataFlow::CallCfgNode {
+  PyOpenSSLContextCreation() {
+    this = API::moduleImport("OpenSSL").getMember("SSL").getMember("Context").getACall()
+  }
+
+  override string getProtocol() {
+    exists(ControlFlowNode protocolArg, PyOpenSSL pyo |
+      protocolArg in [node.getArg(0), node.getArgByName("method")]
+    |
+      protocolArg =
+        [pyo.specific_version(result).getAUse(), pyo.unspecific_version(result).getAUse()]
+            .asCfgNode()
+    )
+  }
+}
+
+class ConnectionCall extends ConnectionCreation, DataFlow::CallCfgNode {
+  ConnectionCall() {
+    this = API::moduleImport("OpenSSL").getMember("SSL").getMember("Connection").getACall()
+  }
+
+  override DataFlow::CfgNode getContext() {
+    result.getNode() in [node.getArg(0), node.getArgByName("context")]
+  }
+}
+
+// This cannot be used to unrestrict,
+// see https://www.pyopenssl.org/en/stable/api/ssl.html#OpenSSL.SSL.Context.set_options
+class SetOptionsCall extends ProtocolRestriction, DataFlow::CallCfgNode {
+  SetOptionsCall() { node.getFunction().(AttrNode).getName() = "set_options" }
+
+  override DataFlow::CfgNode getContext() {
+    result.getNode() = node.getFunction().(AttrNode).getObject()
+  }
+
+  override ProtocolVersion getRestriction() {
+    API::moduleImport("OpenSSL").getMember("SSL").getMember("OP_NO_" + result).getAUse().asCfgNode() in [
+        node.getArg(0), node.getArgByName("options")
+      ]
+  }
+}
+
+class UnspecificPyOpenSSLContextCreation extends PyOpenSSLContextCreation, UnspecificContextCreation {
+  UnspecificPyOpenSSLContextCreation() { library instanceof PyOpenSSL }
+}
+
+class PyOpenSSL extends TlsLibrary {
+  PyOpenSSL() { this = "pyOpenSSL" }
+
+  override string specific_version_name(ProtocolVersion version) { result = version + "_METHOD" }
+
+  override string unspecific_version_name(ProtocolFamily family) {
+    // `"TLS_METHOD"` is not actually available in pyOpenSSL yet, but should be coming soon..
+    result = family + "_METHOD"
+  }
+
+  override API::Node version_constants() { result = API::moduleImport("OpenSSL").getMember("SSL") }
+
+  override ContextCreation default_context_creation() { none() }
+
+  override ContextCreation specific_context_creation() {
+    result instanceof PyOpenSSLContextCreation
+  }
+
+  override DataFlow::Node insecure_connection_creation(ProtocolVersion version) { none() }
+
+  override ConnectionCreation connection_creation() { result instanceof ConnectionCall }
+
+  override ProtocolRestriction protocol_restriction() { result instanceof SetOptionsCall }
+
+  override ProtocolUnrestriction protocol_unrestriction() {
+    result instanceof UnspecificPyOpenSSLContextCreation
+  }
+}