Merge pull request #1272 from GaloisInc/rpc/synchronous-interface

[RPC] Synchronous, typed interface for Python client
This commit is contained in:
Matthew Yacavone 2021-09-17 15:22:23 -04:00 committed by GitHub
commit ee8cbace8f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 683 additions and 187 deletions

View File

@ -18,9 +18,15 @@ from . import solver
from .bitvector import BV
from .commands import *
from .connection import *
# import everything from `.synchronous` except `connect` and `connect_stdio`
# (since functions with those names were already imported from `.connection`)
from .synchronous import Qed, Safe, Counterexample, Satisfiable, Unsatisfiable, CryptolSyncConnection
# and add an alias `sync` for the `synchronous` module
from . import synchronous
sync = synchronous
__all__ = ['bitvector', 'commands', 'connections', 'cryptoltypes', 'solver']
__all__ = ['bitvector', 'commands', 'connection', 'cryptoltypes', 'opaque', 'solver', 'synchronous']
def fail_with(x : Exception) -> NoReturn:

View File

@ -2,9 +2,10 @@
from __future__ import annotations
import base64
from abc import ABC
from enum import Enum
from dataclasses import dataclass
from typing import Any, List, Optional, Union
from typing import Any, Tuple, List, Dict, Optional, Union
from typing_extensions import Literal
import argo_client.interaction as argo
@ -20,7 +21,9 @@ def extend_hex(string : str) -> str:
else:
return string
def from_cryptol_arg(val : Any) -> Any:
CryptolValue = Union[bool, int, BV, Tuple, List, Dict, OpaqueValue]
def from_cryptol_arg(val : Any) -> CryptolValue:
"""Return the canonical Python value for a Cryptol JSON value."""
if isinstance(val, bool):
return val
@ -80,27 +83,36 @@ class CryptolExtendSearchPath(argo.Command):
return res
class CryptolEvalExpr(argo.Command):
class CryptolEvalExprRaw(argo.Command):
def __init__(self, connection : HasProtocolState, expr : Any) -> None:
super(CryptolEvalExpr, self).__init__(
super(CryptolEvalExprRaw, self).__init__(
'evaluate expression',
{'expression': expr},
connection
)
def process_result(self, res : Any) -> Any:
return from_cryptol_arg(res['value'])
return res['value']
class CryptolCall(argo.Command):
class CryptolEvalExpr(CryptolEvalExprRaw):
def process_result(self, res : Any) -> Any:
return from_cryptol_arg(super(CryptolEvalExpr, self).process_result(res))
class CryptolCallRaw(argo.Command):
def __init__(self, connection : HasProtocolState, fun : str, args : List[Any]) -> None:
super(CryptolCall, self).__init__(
super(CryptolCallRaw, self).__init__(
'call',
{'function': fun, 'arguments': args},
connection
)
def process_result(self, res : Any) -> Any:
return from_cryptol_arg(res['value'])
return res['value']
class CryptolCall(CryptolCallRaw):
def process_result(self, res : Any) -> Any:
return from_cryptol_arg(super(CryptolCall, self).process_result(res))
@dataclass
@ -111,43 +123,53 @@ class CheckReport:
error_msg: Optional[str]
tests_run: int
tests_possible: Optional[int]
def __bool__(self) -> bool:
return self.success
class CryptolCheck(argo.Command):
def to_check_report(res : Any) -> CheckReport:
if res['result'] == 'pass':
return CheckReport(
success=True,
args=[],
error_msg = None,
tests_run=res['tests run'],
tests_possible=res['tests possible'])
elif res['result'] == 'fail':
return CheckReport(
success=False,
args=[from_cryptol_arg(arg['expr']) for arg in res['arguments']],
error_msg = None,
tests_run=res['tests run'],
tests_possible=res['tests possible'])
elif res['result'] == 'error':
return CheckReport(
success=False,
args=[from_cryptol_arg(arg['expr']) for arg in res['arguments']],
error_msg = res['error message'],
tests_run=res['tests run'],
tests_possible=res['tests possible'])
else:
raise ValueError("Unknown check result " + str(res))
class CryptolCheckRaw(argo.Command):
def __init__(self, connection : HasProtocolState, expr : Any, num_tests : Union[Literal['all'],int, None]) -> None:
if num_tests:
args = {'expression': expr, 'number of tests':num_tests}
else:
args = {'expression': expr}
super(CryptolCheck, self).__init__(
super(CryptolCheckRaw, self).__init__(
'check',
args,
connection
)
def process_result(self, res : Any) -> Any:
return res
class CryptolCheck(CryptolCheckRaw):
def process_result(self, res : Any) -> 'CheckReport':
if res['result'] == 'pass':
return CheckReport(
success=True,
args=[],
error_msg = None,
tests_run=res['tests run'],
tests_possible=res['tests possible'])
elif res['result'] == 'fail':
return CheckReport(
success=False,
args=[from_cryptol_arg(arg['expr']) for arg in res['arguments']],
error_msg = None,
tests_run=res['tests run'],
tests_possible=res['tests possible'])
elif res['result'] == 'error':
return CheckReport(
success=False,
args=[from_cryptol_arg(arg['expr']) for arg in res['arguments']],
error_msg = res['error message'],
tests_run=res['tests run'],
tests_possible=res['tests possible'])
else:
raise ValueError("Unknown check result " + str(res))
return to_check_report(super(CryptolCheck, self).process_result(res))
class CryptolCheckType(argo.Command):
@ -161,14 +183,15 @@ class CryptolCheckType(argo.Command):
def process_result(self, res : Any) -> Any:
return res['type schema']
class SmtQueryType(str, Enum):
PROVE = 'prove'
SAFE = 'safe'
SAT = 'sat'
class CryptolProveSat(argo.Command):
class CryptolProveSatRaw(argo.Command):
def __init__(self, connection : HasProtocolState, qtype : SmtQueryType, expr : Any, solver : Solver, count : Optional[int]) -> None:
super(CryptolProveSat, self).__init__(
super(CryptolProveSatRaw, self).__init__(
'prove or satisfy',
{'query type': qtype,
'expression': expr,
@ -180,6 +203,11 @@ class CryptolProveSat(argo.Command):
self.qtype = qtype
def process_result(self, res : Any) -> Any:
return res
class CryptolProveSat(CryptolProveSatRaw):
def process_result(self, res : Any) -> Any:
res = super(CryptolProveSat, self).process_result(res)
if res['result'] == 'unsatisfiable':
if self.qtype == SmtQueryType.SAT:
return False
@ -199,18 +227,28 @@ class CryptolProveSat(argo.Command):
else:
raise ValueError("Unknown SMT result: " + str(res))
class CryptolProveRaw(CryptolProveSatRaw):
def __init__(self, connection : HasProtocolState, expr : Any, solver : Solver) -> None:
super(CryptolProveRaw, self).__init__(connection, SmtQueryType.PROVE, expr, solver, 1)
class CryptolProve(CryptolProveSat):
def __init__(self, connection : HasProtocolState, expr : Any, solver : Solver) -> None:
super(CryptolProve, self).__init__(connection, SmtQueryType.PROVE, expr, solver, 1)
class CryptolSatRaw(CryptolProveSatRaw):
def __init__(self, connection : HasProtocolState, expr : Any, solver : Solver, count : int) -> None:
super(CryptolSatRaw, self).__init__(connection, SmtQueryType.SAT, expr, solver, count)
class CryptolSat(CryptolProveSat):
def __init__(self, connection : HasProtocolState, expr : Any, solver : Solver, count : int) -> None:
super(CryptolSat, self).__init__(connection, SmtQueryType.SAT, expr, solver, count)
class CryptolSafeRaw(CryptolProveSatRaw):
def __init__(self, connection : HasProtocolState, expr : Any, solver : Solver) -> None:
super(CryptolSafeRaw, self).__init__(connection, SmtQueryType.SAFE, expr, solver, 1)
class CryptolSafe(CryptolProveSat):
def __init__(self, connection : HasProtocolState, expr : Any, solver : Solver) -> None:
super(CryptolSafe, self).__init__(connection, SmtQueryType.SAFE, expr, solver, 1)
class CryptolNames(argo.Command):
def __init__(self, connection : HasProtocolState) -> None:
super(CryptolNames, self).__init__('visible names', {}, connection)
@ -218,6 +256,7 @@ class CryptolNames(argo.Command):
def process_result(self, res : Any) -> Any:
return res
class CryptolFocusedModule(argo.Command):
def __init__(self, connection : HasProtocolState) -> None:
super(CryptolFocusedModule, self).__init__(
@ -229,6 +268,7 @@ class CryptolFocusedModule(argo.Command):
def process_result(self, res : Any) -> Any:
return res
class CryptolReset(argo.Notification):
def __init__(self, connection : HasProtocolState) -> None:
super(CryptolReset, self).__init__(
@ -237,6 +277,7 @@ class CryptolReset(argo.Notification):
connection
)
class CryptolResetServer(argo.Notification):
def __init__(self, connection : HasProtocolState) -> None:
super(CryptolResetServer, self).__init__(

View File

@ -175,6 +175,13 @@ class CryptolConnection:
self.most_recent_result = CryptolLoadModule(self, module_name)
return self.most_recent_result
def eval_raw(self, expression : Any) -> argo.Command:
"""Like the member method ``eval``, but does not call
``from_cryptol_arg`` on the ``.result()``.
"""
self.most_recent_result = CryptolEvalExprRaw(self, expression)
return self.most_recent_result
def eval(self, expression : Any) -> argo.Command:
"""Evaluate a Cryptol expression, represented according to
:ref:`cryptol-json-expression`, with Python datatypes standing
@ -189,15 +196,33 @@ class CryptolConnection:
return self.eval(expression)
def extend_search_path(self, *dir : str) -> argo.Command:
"""Load a Cryptol module, like ``:module`` at the Cryptol REPL."""
"""Extend the search path for loading Cryptol modules."""
self.most_recent_result = CryptolExtendSearchPath(self, list(dir))
return self.most_recent_result
def call_raw(self, fun : str, *args : List[Any]) -> argo.Command:
"""Like the member method ``call``, but does not call
``from_cryptol_arg`` on the ``.result()``.
"""
encoded_args = [cryptoltypes.CryptolType().from_python(a) for a in args]
self.most_recent_result = CryptolCallRaw(self, fun, encoded_args)
return self.most_recent_result
def call(self, fun : str, *args : List[Any]) -> argo.Command:
encoded_args = [cryptoltypes.CryptolType().from_python(a) for a in args]
self.most_recent_result = CryptolCall(self, fun, encoded_args)
return self.most_recent_result
def check_raw(self, expr : Any, *, num_tests : Union[Literal['all'], int, None] = None) -> argo.Command:
"""Like the member method ``check``, but does not call
`to_check_report` on the ``.result()``.
"""
if num_tests == "all" or isinstance(num_tests, int) or num_tests is None:
self.most_recent_result = CryptolCheckRaw(self, expr, num_tests)
return self.most_recent_result
else:
raise ValueError('``num_tests`` must be an integer, ``None``, or the string literall ``"all"``')
def check(self, expr : Any, *, num_tests : Union[Literal['all'], int, None] = None) -> argo.Command:
"""Tests the validity of a Cryptol expression with random inputs. The expression must be a function with
return type ``Bit``.
@ -212,7 +237,6 @@ class CryptolConnection:
else:
raise ValueError('``num_tests`` must be an integer, ``None``, or the string literall ``"all"``')
def check_type(self, code : Any) -> argo.Command:
"""Check the type of a Cryptol expression, represented according to
:ref:`cryptol-json-expression`, with Python datatypes standing for
@ -221,6 +245,13 @@ class CryptolConnection:
self.most_recent_result = CryptolCheckType(self, code)
return self.most_recent_result
def sat_raw(self, expr : Any, solver : solver.Solver = solver.Z3, count : int = 1) -> argo.Command:
"""Like the member method ``sat``, but does not call
`to_smt_query_result` on the ``.result()``.
"""
self.most_recent_result = CryptolSatRaw(self, expr, solver, count)
return self.most_recent_result
def sat(self, expr : Any, solver : solver.Solver = solver.Z3, count : int = 1) -> argo.Command:
"""Check the satisfiability of a Cryptol expression, represented according to
:ref:`cryptol-json-expression`, with Python datatypes standing for
@ -230,6 +261,13 @@ class CryptolConnection:
self.most_recent_result = CryptolSat(self, expr, solver, count)
return self.most_recent_result
def prove_raw(self, expr : Any, solver : solver.Solver = solver.Z3) -> argo.Command:
"""Like the member method ``prove``, but does not call
`to_smt_query_result` on the ``.result()``.
"""
self.most_recent_result = CryptolProveRaw(self, expr, solver)
return self.most_recent_result
def prove(self, expr : Any, solver : solver.Solver = solver.Z3) -> argo.Command:
"""Check the validity of a Cryptol expression, represented according to
:ref:`cryptol-json-expression`, with Python datatypes standing for
@ -238,6 +276,13 @@ class CryptolConnection:
self.most_recent_result = CryptolProve(self, expr, solver)
return self.most_recent_result
def safe_raw(self, expr : Any, solver : solver.Solver = solver.Z3) -> argo.Command:
"""Like the member method ``safe``, but does not call
`to_smt_query_result` on the ``.result()``.
"""
self.most_recent_result = CryptolSafeRaw(self, expr, solver)
return self.most_recent_result
def safe(self, expr : Any, solver : solver.Solver = solver.Z3) -> argo.Command:
"""Check via an external SMT solver that the given term is safe for all inputs,
which means it cannot encounter a run-time error.

View File

@ -1,14 +1,23 @@
"""Cryptol solver-related definitions"""
from typing import NewType
from abc import ABCMeta, abstractmethod
from typing import Any
from dataclasses import dataclass
@dataclass
class OfflineSmtQuery():
"""An SMTLIB2 script -- produced when an `offline` prover is used."""
"""An SMTLIB2 script -- produced when an `offline` prover is used. Instances
of this class are neither truthy nor falsy, using an instance of this class
in an 'if' or 'while' statement will result in an error.
"""
content: str
class Solver():
def __bool__(self) -> Any:
raise ValueError("Offline SMT query has no value")
def __nonzero__(self) -> Any:
raise ValueError("Offline SMT query has no value")
class Solver(metaclass=ABCMeta):
"""An SMT solver mode selectable for Cryptol. Compare with the `:set prover` options in
the Cryptol REPL."""
__name: str
@ -27,31 +36,40 @@ class Solver():
"""Returns whether hash consing is enabled (if applicable)."""
return self.__hash_consing
@abstractmethod
def without_hash_consing(self) -> 'Solver':
"""Returns an identical solver but with hash consing disabled (if applicable)."""
return Solver(name=self.__name, hash_consing=False)
pass
class OnlineSolver(Solver):
def without_hash_consing(self) -> 'OnlineSolver':
return OnlineSolver(name=self.name(), hash_consing=False)
class OfflineSolver(Solver):
def without_hash_consing(self) -> 'OfflineSolver':
return OfflineSolver(name=self.name(), hash_consing=False)
# Cryptol-supported SMT configurations/solvers
# (see Cryptol.Symbolic.SBV Haskell module)
CVC4: Solver = Solver("cvc4")
YICES: Solver = Solver("yices")
Z3: Solver = Solver("z3")
BOOLECTOR: Solver = Solver("boolector")
MATHSAT: Solver = Solver("mathsat")
ABC: Solver = Solver("abc")
OFFLINE: Solver = Solver("offline")
ANY: Solver = Solver("any")
SBV_CVC4: Solver = Solver("sbv-cvc4")
SBV_YICES: Solver = Solver("sbv-yices")
SBV_Z3: Solver = Solver("sbv-z3")
SBV_BOOLECTOR: Solver = Solver("sbv-boolector")
SBV_MATHSAT: Solver = Solver("sbv-mathsat")
SBV_ABC: Solver = Solver("sbv-abc")
SBV_OFFLINE: Solver = Solver("sbv-offline")
SBV_ANY: Solver = Solver("sbv-any")
W4_CVC4: Solver = Solver("w4-cvc4")
W4_YICES: Solver = Solver("w4-yices")
W4_Z3: Solver = Solver("w4-z3")
W4_BOOLECTOR: Solver = Solver("w4-boolector")
W4_OFFLINE: Solver = Solver("w4-offline")
W4_ABC: Solver = Solver("w4-any")
CVC4: OnlineSolver = OnlineSolver("cvc4")
YICES: OnlineSolver = OnlineSolver("yices")
Z3: OnlineSolver = OnlineSolver("z3")
BOOLECTOR: OnlineSolver = OnlineSolver("boolector")
MATHSAT: OnlineSolver = OnlineSolver("mathsat")
ABC: OnlineSolver = OnlineSolver("abc")
OFFLINE: OfflineSolver = OfflineSolver("offline")
ANY: OnlineSolver = OnlineSolver("any")
SBV_CVC4: OnlineSolver = OnlineSolver("sbv-cvc4")
SBV_YICES: OnlineSolver = OnlineSolver("sbv-yices")
SBV_Z3: OnlineSolver = OnlineSolver("sbv-z3")
SBV_BOOLECTOR: OnlineSolver = OnlineSolver("sbv-boolector")
SBV_MATHSAT: OnlineSolver = OnlineSolver("sbv-mathsat")
SBV_ABC: OnlineSolver = OnlineSolver("sbv-abc")
SBV_OFFLINE: OfflineSolver = OfflineSolver("sbv-offline")
SBV_ANY: OnlineSolver = OnlineSolver("sbv-any")
W4_CVC4: OnlineSolver = OnlineSolver("w4-cvc4")
W4_YICES: OnlineSolver = OnlineSolver("w4-yices")
W4_Z3: OnlineSolver = OnlineSolver("w4-z3")
W4_BOOLECTOR: OnlineSolver = OnlineSolver("w4-boolector")
W4_OFFLINE: OfflineSolver = OfflineSolver("w4-offline")
W4_ABC: OnlineSolver = OnlineSolver("w4-any")

View File

@ -0,0 +1,321 @@
"""A synchronous, typed interface for the Cryptol bindings"""
from __future__ import annotations
import sys
from typing import Any, Optional, Union, List, Dict, TextIO, overload
from typing_extensions import Literal
from dataclasses import dataclass
from .solver import OfflineSmtQuery, Solver, OnlineSolver, OfflineSolver, Z3
from . import connection
from . import cryptoltypes
from .commands import *
from . import CryptolConnection, SmtQueryType
@dataclass
class Qed:
"""The positive result of a 'prove' SMT query. All instances of this class
are truthy, i.e. evaluate to `True` in an 'if' or 'while' statement.
"""
def __bool__(self) -> Literal[True]:
return True
def __nonzero__(self) -> Literal[True]:
return True
@dataclass
class Safe:
"""The positive result of a 'safe' SMT query. All instances of this class
are truthy, i.e. evaluate to `True` in an 'if' or 'while' statement.
"""
def __bool__(self) -> Literal[True]:
return True
def __nonzero__(self) -> Literal[True]:
return True
@dataclass
class Counterexample:
"""The negative result of a 'prove' or 'safe' SMT query, consisting of a
type (either "predicate falsified" or "safety violation") and the list of
values which constitute the counterexample. All instances of this class are
falsy, i.e. evaluate to `False` in an 'if' or 'while' statement. (Note that
this is different from the behaivor of a plain list, which is truthy iff
it has nonzero length.)
"""
type : Union[Literal["predicate falsified"], Literal["safety violation"]]
assignments : List[CryptolValue]
def __bool__(self) -> Literal[False]:
return False
def __nonzero__(self) -> Literal[False]:
return False
@dataclass
class Satisfiable:
"""The positive result of a 'sat' SMT query, consisting of a list of
models, where each model is a list of values satisfying the predicate.
All instances of this class are truthy, i.e. evaluate to `True` in an 'if'
or 'while' statement. (Note that this is different from the behaivor of a
plain list, which is truthy iff it has nonzero length.)
"""
models : List[List[CryptolValue]]
def __bool__(self) -> Literal[True]:
return True
def __nonzero__(self) -> Literal[True]:
return True
@dataclass
class Unsatisfiable:
"""The negative result of a 'sat' SMT query. All instances of this class
are falsy, i.e. evaluate to `False` in an 'if 'or 'while' statement.
"""
def __bool__(self) -> Literal[False]:
return False
def __nonzero__(self) -> Literal[False]:
return False
def connect(command : Optional[str]=None,
*,
cryptol_path : Optional[str] = None,
url : Optional[str] = None,
reset_server : bool = False,
verify : Union[bool, str] = True,
log_dest : Optional[TextIO] = None) -> CryptolSyncConnection:
"""
Connect to a (possibly new) synchronous Cryptol server process.
:param command: A command to launch a new Cryptol server in socket mode (if provided).
:param cryptol_path: A replacement for the contents of
the ``CRYPTOLPATH`` environment variable (if provided).
:param url: A URL at which to connect to an already running Cryptol
HTTP server.
:param reset_server: If ``True``, the server that is connected to will be
reset. (This ensures any states from previous server usages have been
cleared.)
:param verify: Determines whether a secure HTTP connection should verify the SSL certificates.
Corresponds to the ``verify`` keyword parameter on ``requests.post``. N.B.,
only has an affect when ``connect`` is called with a ``url`` parameter
or when the ``CRYPTOL_SERVER_URL`` environment variable is set.
:param log_dest: A destination to log JSON requests/responses to, e.g. ``log_dest=sys.stderr``
will print traffic to ``stderr``, ``log_dest=open('foo.log', 'w')`` will log to ``foo.log``,
etc.
If no ``command`` or ``url`` parameters are provided, the following are attempted in order:
1. If the environment variable ``CRYPTOL_SERVER`` is set and referse to an executable,
it is assumed to be a Cryptol server and will be used for a new ``socket`` connection.
2. If the environment variable ``CRYPTOL_SERVER_URL`` is set, it is assumed to be
the URL for a running Cryptol server in ``http`` mode and will be connected to.
3. If an executable ``cryptol-remote-api`` is available on the ``PATH``
it is assumed to be a Cryptol server and will be used for a new ``socket`` connection.
"""
return CryptolSyncConnection(connection.connect(
command=command,
cryptol_path=cryptol_path,
url=url,
reset_server=reset_server,
verify=verify,
log_dest=log_dest))
def connect_stdio(command : str, cryptol_path : Optional[str] = None) -> CryptolSyncConnection:
"""Start a new synchronous connection to a new Cryptol server process.
:param command: The command to launch the Cryptol server.
:param cryptol_path: An optional replacement for the contents of
the ``CRYPTOLPATH`` environment variable.
"""
return CryptolSyncConnection(connection.connect_stdio(
command=command,
cryptol_path=cryptol_path))
class CryptolSyncConnection:
"""A wrapper of ``CryptolConnection`` with a synchronous, typed interface."""
connection : CryptolConnection
def __init__(self, connection : CryptolConnection):
self.connection = connection
def load_file(self, filename : str) -> None:
"""Load a filename as a Cryptol module, like ``:load`` at the Cryptol
REPL.
"""
self.connection.load_file(filename).result()
def load_module(self, module_name : str) -> None:
"""Load a Cryptol module, like ``:module`` at the Cryptol REPL."""
self.connection.load_module(module_name).result()
def extend_search_path(self, *dir : str) -> None:
"""Extend the search path for loading Cryptol modules."""
self.connection.extend_search_path(*dir).result()
def eval(self, expression : Any) -> CryptolValue:
"""Evaluate a Cryptol expression, with the result represented
according to :ref:`cryptol-json-expression`, with Python datatypes
standing for their JSON equivalents.
"""
return from_cryptol_arg(self.connection.eval_raw(expression).result())
def call(self, fun : str, *args : List[Any]) -> CryptolValue:
"""Evaluate a Cryptol functiom by name, with the arguments and the
result represented according to :ref:`cryptol-json-expression`, with
Python datatypes standing for their JSON equivalents.
"""
return from_cryptol_arg(self.connection.call_raw(fun, *args).result())
def check(self, expr : Any, *, num_tests : Union[Literal['all'], int, None] = None) -> CheckReport:
"""Tests the validity of a Cryptol expression with random inputs. The expression must be a function with
return type ``Bit``.
If ``num_tests`` is ``"all"`` then the expression is tested exhaustively (i.e., against all possible inputs).
If ``num_tests`` is omitted, Cryptol defaults to running 100 tests.
"""
return to_check_report(self.connection.check_raw(expr, num_tests=num_tests).result())
def check_type(self, code : Any) -> cryptoltypes.CryptolType:
"""Check the type of a Cryptol expression, represented according to
:ref:`cryptol-json-expression`, with Python datatypes standing for
their JSON equivalents.
"""
return cryptoltypes.to_type(self.connection.check_type(code).result()['type'])
@overload
def sat(self, expr : Any, solver : OnlineSolver = Z3, count : int = 1) -> Union[Satisfiable, Unsatisfiable]: ...
@overload
def sat(self, expr : Any, solver : OfflineSolver, count : int = 1) -> OfflineSmtQuery: ...
def sat(self, expr : Any, solver : Solver = Z3, count : int = 1) -> Union[Satisfiable, Unsatisfiable, OfflineSmtQuery]:
"""Check the satisfiability of a Cryptol expression, represented according to
:ref:`cryptol-json-expression`, with Python datatypes standing for
their JSON equivalents. Use the solver named `solver`, and return up to
`count` solutions.
If the given solver is an `OnlineSolver`, the result is either an
instance of `Satisfiable`, which is always truthy, or `Unsatisfiable`,
which is always falsy - meaning the result will evaluate to `True` in
an 'if' or 'while' statement iff the given expression is satisfiable.
If the given solver is an `OfflineSolver`, an error will be raised if
the result is used in an 'if' or 'while' statement.
"""
if isinstance(solver, OfflineSolver):
res = self.connection.sat_raw(expr, solver, count).result()
if res['result'] == 'offline':
return OfflineSmtQuery(res['query'])
else:
raise ValueError("Expected an offline SMT result, got: " + str(res))
elif isinstance(solver, OnlineSolver):
res = self.connection.sat_raw(expr, solver, count).result()
if res['result'] == 'unsatisfiable':
return Unsatisfiable()
elif res['result'] == 'satisfied':
return Satisfiable([[from_cryptol_arg(arg['expr']) for arg in m] for m in res['models']])
else:
raise ValueError("Unexpected 'sat' SMT result: " + str(res))
else:
raise ValueError("Unknown solver type: " + str(solver))
@overload
def prove(self, expr : Any, solver : OnlineSolver = Z3) -> Union[Qed, Counterexample]: ...
@overload
def prove(self, expr : Any, solver : OfflineSolver) -> OfflineSmtQuery: ...
def prove(self, expr : Any, solver : Solver = Z3) -> Union[Qed, Counterexample, OfflineSmtQuery]:
"""Check the validity of a Cryptol expression, represented according to
:ref:`cryptol-json-expression`, with Python datatypes standing for
their JSON equivalents. Use the solver named `solver`.
If the given solver is an `OnlineSolver`, the result is either an
instance of `Qed`, which is always truthy, or `Counterexample`, which
is always falsy - meaning the result will evaluate to `True` in an 'if'
or 'while' statement iff the given expression can be proved. If the
given solver is an `OfflineSolver`, an error will be raised if the
result is used in an 'if' or 'while' statement.
"""
if isinstance(solver, OfflineSolver):
res = self.connection.prove_raw(expr, solver).result()
if res['result'] == 'offline':
return OfflineSmtQuery(res['query'])
else:
raise ValueError("Expected an offline SMT result, got: " + str(res))
elif isinstance(solver, OnlineSolver):
res = self.connection.prove_raw(expr, solver).result()
if res['result'] == 'unsatisfiable':
return Qed()
elif res['result'] == 'invalid':
return Counterexample(res['counterexample type'], [from_cryptol_arg(arg['expr']) for arg in res['counterexample']])
else:
raise ValueError("Unexpected 'prove' SMT result: " + str(res))
else:
raise ValueError("Unknown solver type: " + str(solver))
@overload
def safe(self, expr : Any, solver : OnlineSolver = Z3) -> Union[Safe, Counterexample]: ...
@overload
def safe(self, expr : Any, solver : OfflineSolver) -> OfflineSmtQuery: ...
def safe(self, expr : Any, solver : Solver = Z3) -> Union[Safe, Counterexample, OfflineSmtQuery]:
"""Check via an external SMT solver that the given term is safe for all inputs,
which means it cannot encounter a run-time error.
If the given solver is an `OnlineSolver`, the result is either an
instance of `Safe`, which is always truthy, or `Counterexample`, which
is always falsy - meaning the result will evaluate to `True` in an 'if'
or 'while' statement iff the given expression is safe. If the given
solver is an `OfflineSolver`, an error will be raised if the result is
used in an 'if' or 'while' statement.
"""
if isinstance(solver, OfflineSolver):
res = self.connection.safe_raw(expr, solver).result()
if res['result'] == 'offline':
return OfflineSmtQuery(res['query'])
else:
raise ValueError("Expected an offline SMT result, got: " + str(res))
elif isinstance(solver, OnlineSolver):
res = self.connection.safe_raw(expr, solver).result()
if res['result'] == 'unsatisfiable':
return Safe()
elif res['result'] == 'invalid':
return Counterexample(res['counterexample type'], [from_cryptol_arg(arg['expr']) for arg in res['counterexample']])
else:
raise ValueError("Unexpected 'safe' SMT result: " + str(res))
else:
raise ValueError("Unknown solver type: " + str(solver))
def names(self) -> List[Dict[str,Any]]:
"""Discover the list of names currently in scope in the current context."""
res = self.connection.names().result()
if isinstance(res, list) and all(isinstance(d, dict) and all(isinstance(k, str) for k in d.keys()) for d in res):
return res
else:
raise ValueError("Panic! Result of `names()` is malformed: " + str(res))
def focused_module(self) -> Dict[str,Any]:
"""Returns the name and other information about the currently-focused module."""
res = self.connection.focused_module().result()
if isinstance(res, dict) and all(isinstance(k, str) for k in res.keys()):
return res
else:
raise ValueError("Panic! Result of `focused_module()` is malformed: " + str(res))
def reset(self) -> None:
"""Resets the connection, causing its unique state on the server to be freed (if applicable).
After a reset a connection may be treated as if it were a fresh connection with the server if desired."""
self.connection.reset()
def reset_server(self) -> None:
"""Resets the Cryptol server, clearing all states."""
self.connection.reset_server()
def logging(self, on : bool, *, dest : TextIO = sys.stderr) -> None:
"""Whether to log received and transmitted JSON."""
self.connection.server_connection.logging(on=on,dest=dest)

View File

@ -7,30 +7,30 @@ from cryptol.bitvector import BV
class TestAES(unittest.TestCase):
def test_AES(self):
c = cryptol.connect(verify=False)
c = cryptol.sync.connect(verify=False)
c.load_file(str(Path('tests','cryptol','test-files', 'examples','AES.cry')))
pt = BV(size=128, value=0x3243f6a8885a308d313198a2e0370734)
key = BV(size=128, value=0x2b7e151628aed2a6abf7158809cf4f3c)
ct = c.call("aesEncrypt", (pt, key)).result()
ct = c.call("aesEncrypt", (pt, key))
expected_ct = BV(size=128, value=0x3925841d02dc09fbdc118597196a0b32)
self.assertEqual(ct, expected_ct)
decrypted_ct = c.call("aesDecrypt", (ct, key)).result()
decrypted_ct = c.call("aesDecrypt", (ct, key))
self.assertEqual(pt, decrypted_ct)
pt = BV(size=128, value=0x00112233445566778899aabbccddeeff)
key = BV(size=128, value=0x000102030405060708090a0b0c0d0e0f)
ct = c.call("aesEncrypt", (pt, key)).result()
ct = c.call("aesEncrypt", (pt, key))
expected_ct = BV(size=128, value=0x69c4e0d86a7b0430d8cdb78070b4c55a)
self.assertEqual(ct, expected_ct)
decrypted_ct = c.call("aesDecrypt", (ct, key)).result()
decrypted_ct = c.call("aesDecrypt", (ct, key))
self.assertEqual(pt, decrypted_ct)
self.assertTrue(c.safe("aesEncrypt").result())
self.assertTrue(c.safe("aesDecrypt").result())
self.assertTrue(c.check("AESCorrect").result().success)
self.assertTrue(c.safe("aesEncrypt"))
self.assertTrue(c.safe("aesDecrypt"))
self.assertTrue(c.check("AESCorrect"))
# c.prove("AESCorrect") # probably takes too long for this script...?

View File

@ -7,25 +7,25 @@ from cryptol.bitvector import BV
class TestCplxQ(unittest.TestCase):
def test_CplxQ(self):
c = cryptol.connect(verify=False)
c = cryptol.sync.connect(verify=False)
c.load_file(str(Path('tests','cryptol','test-files', 'CplxQNewtype.cry')))
forty_two = c.eval("fortyTwo").result()
cplx_forty_two1 = c.call("embedQ", forty_two).result()
cplx_forty_two2 = c.eval("CplxQ{ real = 42, imag = 0 }").result()
cplx_two = c.eval("cplxTwo").result()
cplx_forty = c.eval("cplxForty").result()
cplx_forty_two3 = c.call("cplxAdd", cplx_two, cplx_forty).result()
cplx_forty_two4 = c.eval("cplxMul (CplxQ{ real = 21, imag = 0 }) (CplxQ{ real = 2, imag = 0 })").result()
cplx_forty_two5 = c.eval("cplxAdd (CplxQ{ real = 41, imag = 0 }) (CplxQ{ real = 1, imag = 0 })").result()
cplx_forty_two6 = c.eval("CplxQ{ real = 42, imag = 0 }").result()
forty_two = c.eval("fortyTwo")
cplx_forty_two1 = c.call("embedQ", forty_two)
cplx_forty_two2 = c.eval("CplxQ{ real = 42, imag = 0 }")
cplx_two = c.eval("cplxTwo")
cplx_forty = c.eval("cplxForty")
cplx_forty_two3 = c.call("cplxAdd", cplx_two, cplx_forty)
cplx_forty_two4 = c.eval("cplxMul (CplxQ{ real = 21, imag = 0 }) (CplxQ{ real = 2, imag = 0 })")
cplx_forty_two5 = c.eval("cplxAdd (CplxQ{ real = 41, imag = 0 }) (CplxQ{ real = 1, imag = 0 })")
cplx_forty_two6 = c.eval("CplxQ{ real = 42, imag = 0 }")
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two2).result())
self.assertFalse(c.call("cplxEq", cplx_two, cplx_forty_two2).result())
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two3).result())
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two4).result())
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two5).result())
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two6).result())
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two2))
self.assertFalse(c.call("cplxEq", cplx_two, cplx_forty_two2))
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two3))
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two4))
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two5))
self.assertTrue(c.call("cplxEq", cplx_forty_two1, cplx_forty_two6))

View File

@ -8,38 +8,38 @@ from cryptol.bitvector import BV
class TestDES(unittest.TestCase):
def test_DES(self):
c = cryptol.connect(verify=False)
c = cryptol.sync.connect(verify=False)
c.load_file(str(Path('tests','cryptol','test-files','examples','DEStest.cry')))
# we can run the test suite as indended...
# vkres = c.eval('vktest DES').result()
# vkres = c.eval('vktest DES')
# self.assertTrue(all(passed for (_,_,passed) in vkres))
# vtres = c.eval('vttest DES').result()
# vtres = c.eval('vttest DES')
# self.assertTrue(all(passed for (_,_,passed) in vtres))
# kares = c.eval('katest DES').result()
# kares = c.eval('katest DES')
# self.assertTrue(all(passed for (_,_,passed) in kares))
# ...but we can also do it manually, using the python bindings more
def test(key, pt0, ct0):
ct1 = c.call('DES.encrypt', key, pt0).result()
pt1 = c.call('DES.decrypt', key, ct0).result()
ct1 = c.call('DES.encrypt', key, pt0)
pt1 = c.call('DES.decrypt', key, ct0)
self.assertEqual(ct0, ct1)
self.assertEqual(pt0, pt1)
# vktest
vk = c.eval('vk').result()
vk = c.eval('vk')
pt0 = BV(size=64, value=0)
for (key, ct0) in vk:
test(key, pt0, ct0)
# vttest
vt = c.eval('vt').result()
vt = c.eval('vt')
key = BV(size=64, value=0x0101010101010101)
for (pt0, ct0) in vt:
test(key, pt0, ct0)
# katest
ka = c.eval('ka').result()
ka = c.eval('ka')
for (key, pt0, ct0) in ka:
test(key, pt0, ct0)

View File

@ -8,18 +8,18 @@ from cryptol.bitvector import BV
class TestEvenMansour(unittest.TestCase):
def test_EvenMansour(self):
c = cryptol.connect(verify=False)
c = cryptol.sync.connect(verify=False)
c.load_file(str(Path('tests','cryptol','test-files','examples','contrib','EvenMansour.cry')))
F_10_4 = c.eval('F:[10][4]').result()
self.assertTrue(c.call('is_a_permutation', F_10_4).result())
F_10_4 = c.eval('F:[10][4]')
self.assertTrue(c.call('is_a_permutation', F_10_4))
Finv_10_4 = c.eval("F':[10][4]").result()
Finv_10_4 = c.eval("F':[10][4]")
digits = [ BV(size=4, value=i) for i in range(0,10) ]
# ^ the same as: c.eval('[0..9]:[_][4]').result()
self.assertTrue(c.call('is_inverse_permutation', digits, F_10_4, Finv_10_4).result())
# ^ the same as: c.eval('[0..9]:[_][4]')
self.assertTrue(c.call('is_inverse_permutation', digits, F_10_4, Finv_10_4))
self.assertTrue(c.check('E_and_D_are_inverses').result().success)
self.assertTrue(c.check('E_and_D_are_inverses'))
if __name__ == "__main__":

View File

@ -8,24 +8,24 @@ from cryptol.bitvector import BV
class TestSHA256(unittest.TestCase):
def test_SHA256(self):
c = cryptol.connect(verify=False)
c = cryptol.sync.connect(verify=False)
c.load_file(str(Path('tests','cryptol','test-files','examples','param_modules','SHA.cry')))
m1 = CryptolLiteral('"abcdbcdecdefdefgefghfghighijhijkijkljklmklmnlmnomnopnopq"')
j1 = c.call('join', m1).result()
h1 = c.call('sha256', j1).result()
j1 = c.call('join', m1)
h1 = c.call('sha256', j1)
expected_h1 = BV(size=256, value=0x248d6a61d20638b8e5c026930c3e6039a33ce45964ff2167f6ecedd419db06c1)
self.assertEqual(h1, expected_h1)
m2 = CryptolLiteral('""')
j2 = c.call('join', m2).result()
h2 = c.call('sha256', j2).result()
j2 = c.call('join', m2)
h2 = c.call('sha256', j2)
expected_h2 = BV(size=256, value=0xe3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855)
self.assertEqual(h2, expected_h2)
m3 = CryptolLiteral('"abcdefghbcdefghicdefghijdefghijkefghijklfghijklmghijklmnhijklmnoijklmnopjklmnopqklmnopqrlmnopqrsmnopqrstnopqrstu"')
j3 = c.call('join', m3).result()
h3 = c.call('sha256', j3).result()
j3 = c.call('join', m3)
h3 = c.call('sha256', j3)
expected_h3 = BV(size=256, value=0xcf5b16a778af8380036ce59e7b0492370b249b11e8f07a51afac45037afee9d1)
self.assertEqual(h3, expected_h3)

View File

@ -17,7 +17,7 @@ class BasicServerTests(unittest.TestCase):
@classmethod
def setUpClass(self):
self.c = cryptol.connect(verify=False)
self.c = cryptol.sync.connect(verify=False)
@classmethod
def tearDownClass(self):
@ -29,25 +29,25 @@ class BasicServerTests(unittest.TestCase):
c = self.c
c.extend_search_path(str(Path('tests','cryptol','test-files', 'test-subdir')))
c.load_module('Bar').result()
ans1 = c.evaluate_expression("theAnswer").result()
ans2 = c.evaluate_expression("id theAnswer").result()
c.load_module('Bar')
ans1 = c.eval("theAnswer")
ans2 = c.eval("id theAnswer")
self.assertEqual(ans1, ans2)
def test_logging(self):
c = self.c
c.extend_search_path(str(Path('tests','cryptol','test-files', 'test-subdir')))
c.load_module('Bar').result()
c.load_module('Bar')
log_buffer = io.StringIO()
c.logging(on=True, dest=log_buffer)
_ = c.evaluate_expression("theAnswer").result()
_ = c.eval("theAnswer")
contents = log_buffer.getvalue()
print(f'CONTENTS: {contents}', file=sys.stderr)
self.assertEqual(len(contents.strip().splitlines()), 2)
_ = c.evaluate_expression("theAnswer").result()
_ = c.eval("theAnswer")
class BasicLoggingServerTests(unittest.TestCase):
@ -58,7 +58,7 @@ class BasicLoggingServerTests(unittest.TestCase):
@classmethod
def setUpClass(self):
self.log_buffer = io.StringIO()
self.c = cryptol.connect(verify=False, log_dest = self.log_buffer)
self.c = cryptol.sync.connect(verify=False, log_dest = self.log_buffer)
@classmethod
def tearDownClass(self):
@ -69,7 +69,7 @@ class BasicLoggingServerTests(unittest.TestCase):
c = self.c
c.extend_search_path(str(Path('tests','cryptol','test-files', 'test-subdir')))
c.load_module('Bar')
_ = c.evaluate_expression("theAnswer").result()
_ = c.eval("theAnswer")
self.assertEqual(len(self.log_buffer.getvalue().splitlines()), 6)

View File

@ -21,7 +21,7 @@ class CryptolTests(unittest.TestCase):
@classmethod
def setUpClass(self):
self.c = cryptol.connect(verify=False)
self.c = cryptol.sync.connect(verify=False)
self.c.load_file(str(Path('tests','cryptol','test-files', 'Foo.cry')))
@classmethod
@ -31,13 +31,13 @@ class CryptolTests(unittest.TestCase):
def test_low_level(self):
c = self.c
x_val = c.evaluate_expression("x").result()
x_val = c.eval("x")
self.assertEqual(c.eval("Id::id x").result(), x_val)
self.assertEqual(c.call('Id::id', bytes.fromhex('ff')).result(), BV(8,255))
self.assertEqual(c.eval("Id::id x"), x_val)
self.assertEqual(c.call('Id::id', bytes.fromhex('ff')), BV(8,255))
self.assertEqual(c.call('add', b'\0', b'\1').result(), BV(8,1))
self.assertEqual(c.call('add', bytes.fromhex('ff'), bytes.fromhex('03')).result(), BV(8,2))
self.assertEqual(c.call('add', b'\0', b'\1'), BV(8,1))
self.assertEqual(c.call('add', bytes.fromhex('ff'), bytes.fromhex('03')), BV(8,2))
# AMK: importing cryptol bindings into Python temporarily broken due to linear state usage changes
# in argo approx 1 March 2020
@ -57,89 +57,99 @@ class CryptolTests(unittest.TestCase):
def test_sat_and_prove(self):
c = self.c
# test a single sat model can be returned
rootsOf9 = c.sat('isSqrtOf9').result()
self.assertEqual(len(rootsOf9), 1)
self.assertTrue(int(rootsOf9[0]) ** 2 % 256, 9)
rootsOf9 = c.sat('isSqrtOf9')
self.assertTrue(rootsOf9)
self.assertEqual(len(rootsOf9.models), 1)
self.assertEqual(len(rootsOf9.models[0]), 1)
self.assertTrue(int(rootsOf9.models[0][0]) ** 2 % 256, 9)
# check we can specify the solver
rootsOf9 = c.sat('isSqrtOf9', solver = solver.ANY).result()
self.assertEqual(len(rootsOf9), 1)
self.assertTrue(int(rootsOf9[0]) ** 2 % 256, 9)
rootsOf9 = c.sat('isSqrtOf9', solver = solver.ANY)
self.assertTrue(rootsOf9)
self.assertEqual(len(rootsOf9.models), 1)
self.assertEqual(len(rootsOf9.models[0]), 1)
self.assertTrue(int(rootsOf9.models[0][0]) ** 2 % 256, 9)
# check we can ask for a specific number of results
rootsOf9 = c.sat('isSqrtOf9', count = 3).result()
self.assertEqual(len(rootsOf9), 3)
self.assertEqual([int(root) ** 2 % 256 for root in rootsOf9], [9,9,9])
rootsOf9 = c.sat('isSqrtOf9', count = 3)
self.assertTrue(rootsOf9)
self.assertEqual(len(rootsOf9.models), 3)
for model in rootsOf9.models:
self.assertEqual(len(model), 1)
self.assertTrue(int(model[0]) ** 2 % 256, 9)
# check we can ask for all results
rootsOf9 = c.sat('isSqrtOf9', count = None).result()
self.assertEqual(len(rootsOf9), 4)
self.assertEqual([int(root) ** 2 % 256 for root in rootsOf9], [9,9,9,9])
rootsOf9 = c.sat('isSqrtOf9', count = None)
self.assertTrue(rootsOf9)
self.assertEqual(len(rootsOf9.models), 4)
for model in rootsOf9.models:
self.assertEqual(len(model), 1)
self.assertTrue(int(model[0]) ** 2 % 256, 9)
# check for an unsat condition
self.assertFalse(c.sat('\\x -> isSqrtOf9 x && ~(elem x [3,131,125,253])').result())
self.assertFalse(c.sat('\\x -> isSqrtOf9 x && ~(elem x [3,131,125,253])'))
# check for a valid condition
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]').result())
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]', solver.Z3).result())
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]', solver.W4_Z3).result())
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]', solver.W4_Z3.without_hash_consing()).result())
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]', solver.SBV_Z3).result())
self.assertIsInstance(c.prove('\\(x : [8]) -> x == reverse (reverse x)', solver.OFFLINE).result(), solver.OfflineSmtQuery)
self.assertIsInstance(c.prove('\\(x : [8]) -> x == reverse (reverse x)', solver.SBV_OFFLINE).result(), solver.OfflineSmtQuery)
self.assertIsInstance(c.prove('\\(x : [8]) -> x == reverse (reverse x)', solver.W4_OFFLINE).result(), solver.OfflineSmtQuery)
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]'))
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]', solver.Z3))
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]', solver.W4_Z3))
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]', solver.W4_Z3.without_hash_consing()))
self.assertTrue(c.prove('\\x -> isSqrtOf9 x ==> elem x [3,131,125,253]', solver.SBV_Z3))
self.assertIsInstance(c.prove('\\(x : [8]) -> x == reverse (reverse x)', solver.OFFLINE), solver.OfflineSmtQuery)
self.assertIsInstance(c.prove('\\(x : [8]) -> x == reverse (reverse x)', solver.SBV_OFFLINE), solver.OfflineSmtQuery)
self.assertIsInstance(c.prove('\\(x : [8]) -> x == reverse (reverse x)', solver.W4_OFFLINE), solver.OfflineSmtQuery)
def test_check(self):
c = self.c
res = c.check("\\x -> x==(x:[8])").result()
res = c.check("\\x -> x==(x:[8])")
self.assertTrue(res.success)
self.assertEqual(res.tests_run, 100)
self.assertEqual(res.tests_possible, 256)
self.assertFalse(len(res.args), 0)
self.assertEqual(res.error_msg, None)
res = c.check("\\x -> x==(x:[8])", num_tests=1).result()
res = c.check("\\x -> x==(x:[8])", num_tests=1)
self.assertTrue(res.success)
self.assertEqual(res.tests_run, 1)
self.assertEqual(res.tests_possible, 256)
self.assertEqual(len(res.args), 0)
self.assertEqual(res.error_msg, None)
res = c.check("\\x -> x==(x:[8])", num_tests=42).result()
res = c.check("\\x -> x==(x:[8])", num_tests=42)
self.assertTrue(res.success)
self.assertEqual(res.tests_run, 42)
self.assertEqual(res.tests_possible, 256)
self.assertEqual(len(res.args), 0)
self.assertEqual(res.error_msg, None)
res = c.check("\\x -> x==(x:[8])", num_tests=1000).result()
res = c.check("\\x -> x==(x:[8])", num_tests=1000)
self.assertTrue(res.success)
self.assertEqual(res.tests_run, 256)
self.assertEqual(res.tests_possible, 256)
self.assertEqual(len(res.args), 0)
self.assertEqual(res.error_msg, None)
res = c.check("\\x -> x==(x:[8])", num_tests='all').result()
res = c.check("\\x -> x==(x:[8])", num_tests='all')
self.assertTrue(res.success)
self.assertEqual(res.tests_run, 256)
self.assertEqual(res.tests_possible, 256)
self.assertEqual(len(res.args), 0)
self.assertEqual(res.error_msg, None)
res = c.check("\\x -> x==(x:Integer)", num_tests=1024).result()
res = c.check("\\x -> x==(x:Integer)", num_tests=1024)
self.assertTrue(res.success)
self.assertEqual(res.tests_run, 1024)
self.assertEqual(res.tests_possible, None)
self.assertEqual(len(res.args), 0)
self.assertEqual(res.error_msg, None)
res = c.check("\\x -> (x + 1)==(x:[8])").result()
res = c.check("\\x -> (x + 1)==(x:[8])")
self.assertFalse(res.success)
self.assertEqual(res.tests_possible, 256)
self.assertEqual(len(res.args), 1)
self.assertEqual(res.error_msg, None)
res = c.check("\\x -> (x / 0)==(x:[8])").result()
res = c.check("\\x -> (x / 0)==(x:[8])")
self.assertFalse(res.success)
self.assertEqual(res.tests_possible, 256)
self.assertEqual(len(res.args), 1)
@ -147,24 +157,27 @@ class CryptolTests(unittest.TestCase):
def test_safe(self):
c = self.c
res = c.safe("\\x -> x==(x:[8])").result()
res = c.safe("\\x -> x==(x:[8])")
self.assertTrue(res)
res = c.safe("\\x -> x / (x:[8])").result()
self.assertEqual(res, [BV(size=8, value=0)])
res = c.safe("\\x -> x / (x:[8])")
self.assertFalse(res)
self.assertEqual(res.assignments, [BV(size=8, value=0)])
res = c.safe("\\x -> x / (x:[8])", solver.Z3).result()
self.assertEqual(res, [BV(size=8, value=0)])
res = c.safe("\\x -> x / (x:[8])", solver.Z3)
self.assertFalse(res)
self.assertEqual(res.assignments, [BV(size=8, value=0)])
res = c.safe("\\x -> x / (x:[8])", solver.W4_Z3).result()
self.assertEqual(res, [BV(size=8, value=0)])
res = c.safe("\\x -> x / (x:[8])", solver.W4_Z3)
self.assertFalse(res)
self.assertEqual(res.assignments, [BV(size=8, value=0)])
def test_many_usages_one_connection(self):
c = self.c
for i in range(0,100):
x_val1 = c.evaluate_expression("x").result()
x_val2 = c.eval("Id::id x").result()
x_val1 = c.eval("x")
x_val2 = c.eval("Id::id x")
self.assertEqual(x_val1, x_val2)
@ -216,20 +229,20 @@ class HttpMultiConnectionTests(unittest.TestCase):
def test_reset_with_many_usages_many_connections(self):
for i in range(0,100):
time.sleep(.05)
c = cryptol.connect(url=self.url, verify=False)
c = cryptol.sync.connect(url=self.url, verify=False)
c.load_file(str(Path('tests','cryptol','test-files', 'Foo.cry')))
x_val1 = c.evaluate_expression("x").result()
x_val2 = c.eval("Id::id x").result()
x_val1 = c.eval("x")
x_val2 = c.eval("Id::id x")
self.assertEqual(x_val1, x_val2)
c.reset()
def test_server_with_many_usages_many_connections(self):
for i in range(0,100):
time.sleep(.05)
c = cryptol.connect(url=self.url, verify=False)
c = cryptol.sync.connect(url=self.url, verify=False)
c.load_file(str(Path('tests','cryptol','test-files', 'Foo.cry')))
x_val1 = c.evaluate_expression("x").result()
x_val2 = c.eval("Id::id x").result()
x_val1 = c.eval("x")
x_val2 = c.eval("Id::id x")
self.assertEqual(x_val1, x_val2)
@ -281,10 +294,10 @@ class TLSConnectionTests(unittest.TestCase):
def test_tls_connection(self):
if self.run_tests:
c = cryptol.connect(url=self.url, verify=False)
c = cryptol.sync.connect(url=self.url, verify=False)
c.load_file(str(Path('tests','cryptol','test-files', 'Foo.cry')))
x_val1 = c.evaluate_expression("x").result()
x_val2 = c.eval("Id::id x").result()
x_val1 = c.eval("x")
x_val2 = c.eval("Id::id x")
self.assertEqual(x_val1, x_val2)
if __name__ == "__main__":

View File

@ -9,17 +9,17 @@ from cryptol.bitvector import BV
class TestErrorRecovery(unittest.TestCase):
def test_ErrorRecovery(self):
c = cryptol.connect()
c = cryptol.sync.connect()
with self.assertRaises(ArgoException):
c.load_file(str(Path('tests','cryptol','test-files','bad.cry'))).result()
c.load_file(str(Path('tests','cryptol','test-files','bad.cry')))
# test that loading a valid file still works after an exception
c.load_file(str(Path('tests','cryptol','test-files','Foo.cry'))).result()
c.load_file(str(Path('tests','cryptol','test-files','Foo.cry')))
# ensure that we really did load the file with no errors
x_val1 = c.evaluate_expression("x").result()
x_val2 = c.eval("Id::id x").result()
x_val1 = c.eval("x")
x_val2 = c.eval("Id::id x")
self.assertEqual(x_val1, x_val2)
# test that a reset still works after an exception (this used to throw

View File

@ -0,0 +1,52 @@
import unittest
from pathlib import Path
import unittest
import cryptol
from cryptol.cryptoltypes import CryptolApplication, CryptolLiteral
from cryptol.bitvector import BV
class TestSMT(unittest.TestCase):
def test_SMT(self):
c = cryptol.sync.connect(verify=False)
c.load_module('Cryptol')
ex_true = '\(x : [8]) -> x+x == x+x'
ex_true_safe = c.safe(ex_true)
self.assertTrue(ex_true_safe)
self.assertIsInstance(ex_true_safe, cryptol.Safe)
ex_true_prove = c.prove(ex_true)
self.assertTrue(ex_true_prove)
self.assertIsInstance(ex_true_prove, cryptol.Qed)
ex_true_sat = c.sat(ex_true)
self.assertTrue(ex_true_sat)
self.assertIsInstance(ex_true_sat, cryptol.Satisfiable)
ex_false = '\(x : [8]) -> x+2*x+1 == x'
ex_false_safe = c.safe(ex_false)
self.assertTrue(ex_false_safe)
self.assertIsInstance(ex_false_safe, cryptol.Safe)
ex_false_prove = c.prove(ex_false)
self.assertFalse(ex_false_prove)
self.assertIsInstance(ex_false_prove, cryptol.Counterexample)
self.assertEqual(ex_false_prove.type, "predicate falsified")
ex_false_sat = c.sat(ex_false)
self.assertFalse(ex_false_sat)
self.assertIsInstance(ex_false_sat, cryptol.Unsatisfiable)
ex_partial = '\(x : [8]) -> if x < 0x0f then x == x else error "!"'
ex_partial_safe = c.safe(ex_partial)
self.assertFalse(ex_partial_safe)
self.assertIsInstance(ex_partial_safe, cryptol.Counterexample)
self.assertEqual(ex_partial_safe.type, "safety violation")
ex_partial_prove = c.prove(ex_partial)
self.assertFalse(ex_partial_prove)
self.assertIsInstance(ex_partial_prove, cryptol.Counterexample)
self.assertEqual(ex_partial_prove.type, "safety violation")
ex_partial_sat = c.sat(ex_partial)
self.assertTrue(ex_partial_sat)
self.assertIsInstance(ex_partial_sat, cryptol.Satisfiable)
if __name__ == "__main__":
unittest.main()

View File

@ -8,36 +8,36 @@ from cryptol.bitvector import BV
class CryptolTypeTests(unittest.TestCase):
def test_types(self):
c = cryptol.connect(verify=False)
c = cryptol.sync.connect(verify=False)
c.load_file(str(Path('tests','cryptol','test-files','Types.cry')))
# Bits
self.assertEqual(c.eval('b').result(), True)
self.assertEqual(c.eval('b'), True)
# Words
self.assertEqual(c.eval('w').result(), BV(size=16, value=42))
self.assertEqual(c.eval('w'), BV(size=16, value=42))
# Integers
self.assertEqual(c.eval('z').result(), 420000)
self.assertEqual(c.eval('z'), 420000)
# Modular integers - not supported at all
with self.assertRaises(ValueError):
c.eval('m').result()
c.eval('m')
# Rationals - supported only as opaque values
self.assertIsInstance(c.eval('q').result(), OpaqueValue)
self.assertIsInstance(c.eval('q'), OpaqueValue)
# Tuples
self.assertEqual(c.eval('t').result(), (False, 7))
self.assertEqual(c.eval('t'), (False, 7))
# Sequences
self.assertEqual(c.eval('s').result(),
self.assertEqual(c.eval('s'),
[[BV(size=8, value=1), BV(size=8, value=2), BV(size=8, value=3)],
[BV(size=8, value=4), BV(size=8, value=5), BV(size=8, value=6)],
[BV(size=8, value=7), BV(size=8, value=8), BV(size=8, value=9)]])
# Records
self.assertEqual(c.eval('r').result(),
self.assertEqual(c.eval('r'),
{'xCoord': BV(size=32, value=12),
'yCoord': BV(size=32, value=21)})