Combine options outside of Z3

.github/workflows/actions.yml

@@ -17,4 +17,6 @@ jobs:
           pip install poetry
           poetry install
       - name: Run tests
-        run: 'poetry run make test'
+        run: poetry run make test
+      - name: Run linter
+        run: poetry run make lint-check

predicate/Makefile

@@ -1,16 +1,15 @@
-all:
+all: lint test
 
 .PHONY: test-%
 test-%:
-	mypy solver/test/test_$*.py
 	pytest --pyargs solver/test/test_$*.py
 
 .PHONY: lint
 lint:
 	lint-python
 
-.PHONY: check
-check:
+.PHONY: lint-check
+lint-check:
 	lint-python --check
 
 .PHONY: test

predicate/README.md

@@ -10,83 +10,16 @@ Alternately, `poetry shell` can also be used to run `predicate`.
 
 ## Working with policies
 
-### Example policy
-
-```py
-# access.py
-
-from solver.ast import Duration
-from solver.teleport import AccessNode, Node, Options, OptionsSet, Policy, Rules, User
-
-
-class Teleport:
-    p = Policy(
-        name="access",
-        loud=False,
-        allow=Rules(
-            AccessNode(
-                ((AccessNode.login == User.name) & (User.name != "root"))
-                | (User.traits["team"] == ("admins",))
-            ),
-        ),
-        options=OptionsSet(Options((Options.max_session_ttl < Duration.new(hours=10)))),
-        deny=Rules(
-            AccessNode(
-                (AccessNode.login == "mike")
-                | (AccessNode.login == "jester")
-                | (Node.labels["env"] == "prod")
-            ),
-        ),
-    )
-
-    def test_access(self):
-        # Alice will be able to login to any machine as herself
-        ret, _ = self.p.check(
-            AccessNode(
-                (AccessNode.login == "alice")
-                & (User.name == "alice")
-                & (Node.labels["env"] == "dev")
-            )
-        )
-        assert ret is True, "Alice can login with her user to any node"
-
-        # No one is permitted to login as mike
-        ret, _ = self.p.query(AccessNode((AccessNode.login == "mike")))
-        assert ret is False, "This role does not allow access as mike"
-
-        # No one is permitted to login as jester
-        ret, _ = self.p.query(AccessNode((AccessNode.login == "jester")))
-        assert ret is False, "This role does not allow access as jester"
-```
+See example policies in the [examples/](examples/) folder.
 
 ### Testing a policy
 
 ```bash
-predicate test access.py
-```
-
-```bash
-Running 1 tests:
-  - test_access: ok
+predicate test examples/access.py
 ```
 
 ### Exporting a policy
 
 ```bash
-predicate export access.py
+predicate export examples/access.py
 ```
-
-```yaml
-kind: policy
-metadata:
-  name: access
-spec:
-  allow:
-    access_node: (((access_node.login == user.name) && (!(user.name == "root"))) ||
-      equals(user.traits["team"], ["admins"]))
-  deny:
-    access_node: (((access_node.login == "mike") || (access_node.login == "jester"))
-      || (node.labels["env"] == "prod"))
-  options: (options.max_session_ttl < 36000000000000)
-version: v1
-```

predicate/cli/__main__.py

@@ -5,6 +5,8 @@
 import click
 import yaml
 
+from solver.teleport import Policy
+
 
 @click.group()
 def main():
@@ -17,7 +19,7 @@ def export(policy_file):
     module = run_path(policy_file)
 
     # Grabs the class and directly reads the policy since it's a static member.
-    policy = module["Teleport"].p
+    policy: Policy = module["Teleport"].p
 
     # Dump the policy into a Teleport resource and write it to the terminal.
     obj = policy.export()
@@ -31,10 +33,12 @@ def export(policy_file):
 def deploy(policy_file, sudo):
     click.echo("parsing policy...")
     module = run_path(policy_file)
-    policy = module["Teleport"].p
+    policy: Policy = module["Teleport"].p
+
     click.echo("translating policy...")
     obj = policy.export()
     serialized = yaml.dump(obj)
+
     click.echo("deploying policy...")
     args = ["tctl", "create", "-f"]
     if sudo:

predicate/examples/access.py

@@ -1,18 +1,31 @@
 from solver.ast import Duration
-from solver.teleport import AccessNode, Node, Options, OptionsSet, Policy, Rules, User
+from solver.teleport import (
+    AccessNode,
+    Node,
+    Options,
+    Policy,
+    RecordingMode,
+    Rules,
+    SourceIp,
+    User,
+)
 
 
 class Teleport:
     p = Policy(
         name="access",
         loud=False,
+        options=Options(
+            max_session_ttl=Duration.new(hours=10),
+            recording_mode=RecordingMode.STRICT,
+            source_ip=SourceIp.PINNED,
+        ),
         allow=Rules(
             AccessNode(
                 ((AccessNode.login == User.name) & (User.name != "root"))
                 | (User.traits["team"] == ("admins",))
             ),
         ),
-        options=OptionsSet(Options((Options.max_session_ttl < Duration.new(hours=10)))),
         deny=Rules(
             AccessNode(
                 (AccessNode.login == "mike")
@@ -24,19 +37,19 @@ class Teleport:
 
     def test_access(self):
         # Alice will be able to login to any machine as herself
-        ret, _ = self.p.check(
+        ret = self.p.check(
             AccessNode(
                 (AccessNode.login == "alice")
                 & (User.name == "alice")
                 & (Node.labels["env"] == "dev")
             )
         )
-        assert ret is True, "Alice can login with her user to any node"
+        assert ret.solves is True, "Alice can login with her user to any node"
 
         # No one is permitted to login as mike
-        ret, _ = self.p.query(AccessNode((AccessNode.login == "mike")))
-        assert ret is False, "This role does not allow access as mike"
+        ret = self.p.query(AccessNode((AccessNode.login == "mike")))
+        assert ret.solves is False, "This role does not allow access as mike"
 
         # No one is permitted to login as jester
-        ret, _ = self.p.query(AccessNode((AccessNode.login == "jester")))
-        assert ret is False, "This role does not allow access as jester"
+        ret = self.p.query(AccessNode((AccessNode.login == "jester")))
+        assert ret.solves is False, "This role does not allow access as jester"

predicate/examples/join_session.py

@@ -1,4 +1,4 @@
-from solver.teleport import JoinSession, Session, Policy, Rules, User, AccessNode
+from solver.teleport import JoinSession, Policy, Rules, Session, User
 
 
 class Teleport:
@@ -9,51 +9,58 @@ class Teleport:
             # Equivalent to `join_sessions`:
             # https://goteleport.com/docs/access-controls/guides/moderated-sessions/#join_sessions
             JoinSession(
-                (User.traits["team"].contains("dev")) &
-                ((JoinSession.mode == "observer") | (JoinSession.mode == "peer")) &
-                ((Session.owner.traits["team"].contains("dev")) | (Session.owner.traits["team"].contains("intern")))
+                (User.traits["team"].contains("dev"))
+                & ((JoinSession.mode == "observer") | (JoinSession.mode == "peer"))
+                & (
+                    (Session.owner.traits["team"].contains("dev"))
+                    | (Session.owner.traits["team"].contains("intern"))
+                )
             ),
         ),
-        deny=Rules(
-            JoinSession(
-                User.traits["team"].contains("intern")
-            )
-        )
+        deny=Rules(JoinSession(User.traits["team"].contains("intern"))),
     )
 
     def test_access(self):
-        ret, _ = self.p.check(
+        ret = self.p.check(
             JoinSession(
-                (User.traits["team"] == ("dev",)) &
-                (JoinSession.mode == "observer") &
-                (Session.owner.traits["team"] == ("intern",))
+                (User.traits["team"] == ("dev",))
+                & (JoinSession.mode == "observer")
+                & (Session.owner.traits["team"] == ("intern",))
             )
         )
-        assert ret is True, "a dev user can join a session from an intern user as an observer"
+        assert (
+            ret.solves is True
+        ), "a dev user can join a session from an intern user as an observer"
 
-        ret, _ = self.p.check(
+        ret = self.p.check(
             JoinSession(
-                (User.traits["team"] == ("marketing",)) &
-                (JoinSession.mode == "observer") &
-                (Session.owner.traits["team"] == ("intern",))
+                (User.traits["team"] == ("marketing",))
+                & (JoinSession.mode == "observer")
+                & (Session.owner.traits["team"] == ("intern",))
             )
         )
-        assert ret is False, "a marketing user cannot join a session from an intern user as an observer"
+        assert (
+            ret.solves is False
+        ), "a marketing user cannot join a session from an intern user as an observer"
 
-        ret, _ = self.p.check(
+        ret = self.p.check(
             JoinSession(
-                (User.traits["team"] == ("dev",)) &
-                (JoinSession.mode == "moderator") &
-                (Session.owner.traits["team"] == ("intern",))
+                (User.traits["team"] == ("dev",))
+                & (JoinSession.mode == "moderator")
+                & (Session.owner.traits["team"] == ("intern",))
             )
         )
-        assert ret is False, "a dev user cannot join a session from an intern user as a moderator"
+        assert (
+            ret.solves is False
+        ), "a dev user cannot join a session from an intern user as a moderator"
 
-        ret, _ = self.p.check(
+        ret = self.p.check(
             JoinSession(
-                (User.traits["team"] == ("dev", "intern")) &
-                (JoinSession.mode == "observer") &
-                (Session.owner.traits["team"] == ("intern",))
+                (User.traits["team"] == ("dev", "intern"))
+                & (JoinSession.mode == "observer")
+                & (Session.owner.traits["team"] == ("intern",))
             )
         )
-        assert ret is False, "a dev intern user cannot join a session from an intern user as an observer"
+        assert (
+            ret.solves is False
+        ), "a dev intern user cannot join a session from an intern user as an observer"

predicate/mypy.ini

@@ -1,4 +1,3 @@
 [mypy]
 ignore_missing_imports = True
-exclude = examples