Introduce custom sympy srepr parser [stable/1.4] (#14022)

* Introduce custom sympy srepr parser

This commit introduces a custom parser to QPY for parameter expression
payloads that were generated using sympy. Prior to QPY version 10 this
was the only way we supported serializing parameter expressions in QPY.
For QPY version 10, 11, and 12 sympy could optionally be used if the
payload was generated explicitly to not use symengine (in qiskit 1.0 it
defaulted to use symengine).
This serialization format relied on sympy to generate a string
representation of the expression which we then put in the payload. On
deserialization we called sympy's `parse_expr()` function which
internally is calling sympy's `sympify()` internally. Sympy documents
that `sympify()` relies on Python's `eval()` for string input and
should not be used with untrusted input. But `parse_expr()` didn't have
such a warning (at the time, I plan to contribute adding one), so
using this function provides an avenue for arbitrary code execution
during QPY deserialization.

This commit fixes this issue by writing a custom parser for the string
repesentation in a QPY payload based on python's ast module. This parser
walks the abstract syntax tree and builds the sympy expression object as
it it goes. It is restricted to the operations that
`ParameterExpression` supports and if any part of the string tries to
use functionality outside that set it will error.

For `ScheduleBlock` objects which also serialize a symbolic expression
there is no new parser to securely load the payload. This is because
the data model of `ScheduleBlock` enables using any `sympy` expression
and the limited parser is not sufficient to represent the schedule
block. Since this functionality is deprecated and will be removed in
Qiskit 2.0.0, potentially vulnerable payloads are rejected by default
now. If the input is trusted by the user they can explicitly opt-in to
using it by setting a new flag to trust the input. This is technically a
breaking API change, but since the payload is potentially vulnerable
this is a necessary change to secure QPY.

* Fix calibrations deserialization

The calibrations path was apparently never using symengine encoding for
pulse schedules even if it was available and set. This is a bug but fixing
it prevents us from loading historical payloads. Since this is
pre-existing this restores the previous behavior. We can fix the
underlying bug in an independent PR for a 1.4.x release that uses the
qiskit version as a gate on how to interpret the symbolic encoding flag.

* Suppress lint errors in ast walker

* Set trust flag on qpy tests

* Apply suggestions from code review

Co-authored-by: Jake Lishman <[email protected]>

* Check we don't call eval via sympy ever

This commit looks for all the calls to sympy that internally call eval()
and either does a runtime check to ensure we're not passing user defined
strings to those functions or rewrites the functions to not rely on
those sympy functions.

* Add loud warning to the qpy.load() docs

* Make untrusted sympy exception local to sympy usage

Previously the exception raised in the pulse path was over eagerly
raising for any pulse payload using sympy encoding, whether it contained
a symbolic pulse or not. This commit fixes this oversight so that the
exception is only raised when we go to use sympy's deserialization.
Tests are added to ensure we still raise with a symbolic pulse but not
if the schedule block doesn't contain any symbolic expressions.

* Apply suggestions from code review

Co-authored-by: Elena Peña Tapia <[email protected]>

* Fix trailing whitespace

* Prepare fix as 1.4.2 release

---------

Co-authored-by: Jake Lishman <[email protected]>
Co-authored-by: Elena Peña Tapia <[email protected]>

Author: 3 people

Cargo.toml

@@ -3,7 +3,7 @@ members = ["crates/*"]
 resolver = "2"
 
 [workspace.package]
-version = "1.4.1"
+version = "1.4.2"
 edition = "2021"
 rust-version = "1.70"  # Keep in sync with README.md and rust-toolchain.toml.
 license = "Apache-2.0"

docs/conf.py

@@ -32,7 +32,7 @@
 # The short X.Y version
 version = "1.4"
 # The full version, including alpha/beta/rc tags
-release = "1.4.1"
+release = "1.4.2"
 
 language = "en"
 

qiskit/VERSION.txt

@@ -1 +1 @@
-1.4.1
+1.4.2

qiskit/circuit/parameterexpression.py

@@ -25,7 +25,7 @@
 import numpy
 import symengine
 
-from qiskit.circuit.exceptions import CircuitError
+from qiskit.circuit.exceptions import CircuitError, QiskitError
 
 # This type is redefined at the bottom to insert the full reference to "ParameterExpression", so it
 # can safely be used by runtime type-checkers like Sphinx.  Mypy does not need this because it
@@ -528,6 +528,9 @@ def __repr__(self):
     def __str__(self):
         from sympy import sympify, sstr
 
+        if not isinstance(self._symbol_expr, symengine.Basic):
+            raise QiskitError("Invalid ParameterExpression")
+
         return sstr(sympify(self._symbol_expr), full_prec=False)
 
     def __complex__(self):
@@ -608,6 +611,8 @@ def __eq__(self, other):
                 return False
             from sympy import sympify
 
+            if not isinstance(self._symbol_expr, symengine.Basic):
+                raise QiskitError("Invalid ParameterExpression")
             return sympify(self._symbol_expr).equals(sympify(other._symbol_expr))
         elif isinstance(other, numbers.Number):
             return len(self.parameters) == 0 and complex(self._symbol_expr) == other

qiskit/circuit/tools/pi_check.py

@@ -49,7 +49,10 @@ def pi_check(inpt, eps=1e-9, output="text", ndigits=None):
     if isinstance(inpt, ParameterExpression):
         param_str = str(inpt)
         from sympy import sympify
+        import symengine
 
+        if not isinstance(inpt._symbol_expr, symengine.Basic):
+            raise QiskitError("Invalid ParameterExpression provided")
         expr = sympify(inpt._symbol_expr)
         syms = expr.atoms()
         for sym in syms:

qiskit/qpy/binary_io/circuits.py

@@ -638,7 +638,7 @@ def _read_custom_operations(file_obj, version, vectors):
     return custom_operations
 
 
-def _read_calibrations(file_obj, version, vectors, metadata_deserializer):
+def _read_calibrations(file_obj, version, vectors, metadata_deserializer, trust_input=False):
     calibrations = {}
 
     header = formats.CALIBRATION._make(
@@ -656,7 +656,12 @@ def _read_calibrations(file_obj, version, vectors, metadata_deserializer):
         params = tuple(
             value.read_value(file_obj, version, vectors) for _ in range(defheader.num_params)
         )
-        schedule = schedules.read_schedule_block(file_obj, version, metadata_deserializer)
+        schedule = schedules.read_schedule_block(
+            file_obj,
+            version,
+            metadata_deserializer,
+            trust_input=trust_input,
+        )
 
         if name not in calibrations:
             calibrations[name] = {(qubits, params): schedule}
@@ -1327,7 +1332,9 @@ def write_circuit(
     _write_layout(file_obj, circuit)
 
 
-def read_circuit(file_obj, version, metadata_deserializer=None, use_symengine=False):
+def read_circuit(
+    file_obj, version, metadata_deserializer=None, use_symengine=False, trust_input=False
+):
     """Read a single QuantumCircuit object from the file like object.
 
     Args:
@@ -1345,6 +1352,7 @@ def read_circuit(file_obj, version, metadata_deserializer=None, use_symengine=Fa
             supported in all platforms. Please check that your target platform is supported by
             the symengine library before setting this option, as it will be required by qpy to
             deserialize the payload.
+        trust_input (bool): If true, deserialize vulnerable schedule block payloads.
     Returns:
         QuantumCircuit: The circuit object from the file.
 
@@ -1454,7 +1462,11 @@ def read_circuit(file_obj, version, metadata_deserializer=None, use_symengine=Fa
     # Read calibrations
     if version >= 5:
         circ._calibrations_prop = _read_calibrations(
-            file_obj, version, vectors, metadata_deserializer
+            file_obj,
+            version,
+            vectors,
+            metadata_deserializer,
+            trust_input=trust_input,
         )
 
     for vec_name, (vector, initialized_params) in vectors.items():

qiskit/qpy/binary_io/parse_sympy_repr.py

@@ -0,0 +1,121 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2025.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Parser for sympy expressions srepr from ParameterExpression internals."""
+
+import ast
+
+from qiskit.qpy.exceptions import QpyError
+
+
+ALLOWED_CALLERS = {
+    "Abs",
+    "Add",
+    "Sub",
+    "Mul",
+    "Div",
+    "Pow",
+    "Symbol",
+    "Integer",
+    "Rational",
+    "Complex",
+    "Float",
+    "log",
+    "sin",
+    "cos",
+    "tan",
+    "atan",
+    "acos",
+    "asin",
+    "exp",
+    "conjugate",
+}
+
+UNARY = {
+    "sin",
+    "cos",
+    "tan",
+    "atan",
+    "acos",
+    "asin",
+    "conjugate",
+    "Symbol",
+    "Integer",
+    "Complex",
+    "Abs",
+    "Float",
+}
+
+
+class ParseSympyWalker(ast.NodeVisitor):
+    """A custom ast walker that is passed the sympy srepr from QPY < 13 and creates a custom
+    expression."""
+
+    def __init__(self):
+        self.stack = []
+
+    def visit_UnaryOp(self, node: ast.UnaryOp):  # pylint: disable=invalid-name
+        """Visit a python unary op node"""
+        self.visit(node.operand)
+        arg = self.stack.pop()
+        if isinstance(node.op, ast.UAdd):
+            self.stack.append(+arg)
+        elif isinstance(node.op, ast.USub):
+            self.stack.append(-arg)
+        elif isinstance(node.op, ast.Not):
+            self.stack.append(not arg)
+        elif isinstance(node.op, ast.Invert):
+            self.stack.append(~arg)
+        else:
+            raise QpyError(f"Invalid unary op as part of sympy srepr: {node.op}")
+
+    def visit_Constant(self, node: ast.Constant):  # pylint: disable=invalid-name
+        """Visit a constant node."""
+        self.stack.append(node.value)
+
+    def visit_Call(self, node: ast.Call):  # pylint: disable=invalid-name
+        """Visit a call node
+
+        This can only be parameter expression allowed sympy call types.
+        """
+        import sympy
+
+        if isinstance(node.func, ast.Name):
+            name = node.func.id
+        else:
+            raise QpyError("Unknown node type")
+
+        if name not in ALLOWED_CALLERS:
+            raise QpyError(f"{name} is not part of a valid sympy expression srepr")
+
+        args = node.args
+        if name in UNARY:
+            if len(args) != 1:
+                raise QpyError(f"{name} has an invalid number of args in sympy srepr")
+            self.visit(args[0])
+            obj = getattr(sympy, name)(self.stack.pop())
+            self.stack.append(obj)
+        else:
+            for arg in args:
+                self.visit(arg)
+            out_args = [self.stack.pop() for _ in range(len(args))]
+            out_args.reverse()
+            obj = getattr(sympy, name)(*out_args)
+            self.stack.append(obj)
+
+
+def parse_sympy_repr(sympy_repr: str):
+    """Parse a given sympy srepr into a symbolic expression object."""
+    tree = ast.parse(sympy_repr, mode="eval")
+    visitor = ParseSympyWalker()
+    visitor.visit(tree)
+    return visitor.stack.pop()