adds who can batch/home/skybound/src/cloud/iamspy/.venv

iamspy/cli.py

@@ -82,6 +82,33 @@ def who_can(
     print("\n".join(m.who_can(action, resource, conditions, condition_file, strict_conditions)))
 
 
+@app.command()
+def who_can_batch_resource(
+    action: str = typer.Argument(...),
+    resources_file: str = typer.Argument(...),
+    conditions: List[str] = typer.Option([], "-c", help="List of conditions as key=value string pairs"),
+    condition_file: Optional[str] = typer.Option(
+        None, "-C", help="File of conditions to load following IAM condition syntax"
+    ),
+    strict_conditions: bool = typer.Option(
+        False, help="Whether to require conditions to be passed in for any IAM condition checks"
+    ),
+    model: str = typer.Option("model.smt2", "-f"),
+):
+    """
+    Pulls out applicable policies, runs who_can
+    """
+    m = Model()
+    if Path(model).is_file():
+        m.load_model(model)
+
+    with open(resources_file) as fs:
+        resources = fs.read().strip().split("\n")
+
+    for source, resource in m.who_can_batch_resource(action, resources, conditions, condition_file, strict_conditions):
+        print(f"{source} can perform {action} on {resource}")
+
+
 @app.callback()
 def main(verbose: int = typer.Option(0, "--verbose", "-v", count=True)):
     """

iamspy/iam.py

@@ -92,6 +92,21 @@ class Tag:
     Value: str
 
 
+@dataclass
+class AccessKey:
+    UserName: str
+    AccessKeyId: str
+    Status: str
+    CreateDate: datetime
+
+
+@dataclass
+class LoginProfile:
+    UserName: str
+    CreateDate: datetime
+    PasswordResetRequired: bool
+
+
 @dataclass
 class UserDetail:
     Path: str
@@ -100,8 +115,11 @@ class UserDetail:
     Arn: str
     CreateDate: datetime
     UserPolicyList: List[Policy] = Field(default_factory=list)
+    PasswordLastUsed: Optional[datetime] = None
     GroupList: List[str] = Field(default_factory=list)
     AttachedManagedPolicies: List[ManagedPolicy] = Field(default_factory=list)
+    LoginProfile: Optional[LoginProfile] = None
+    AccessKeys: Optional[List[AccessKey]] = None
     PermissionsBoundary: Optional[PermissionBoundary] = None
     Tags: List[Tag] = Field(default_factory=list)
 

iamspy/model.py

@@ -1,4 +1,4 @@
-from typing import List, Optional, Set
+from typing import List, Optional, Set, Union, Tuple
 import logging
 import json
 import z3
@@ -105,17 +105,20 @@ def model_vars(self) -> Set[str]:
     def hash(self):
         return hashlib.md5(self.solver.to_smt2().encode()).hexdigest()
 
-    def generate_evaluation_logic_checks(self, source: Optional[str], resource: str):
+    def generate_evaluation_logic_checks(self, source: Optional[str], resource: Union[str, List[str]]):
         """
         Generate the assertions for the model
         """
+        if isinstance(resource, str):
+            resource = [resource]
+
         return parse.generate_evaluation_logic_checks(self.model_vars, source, resource)
 
     def _generate_query_conditions(
         self,
         source: Optional[str],
         action: str,
-        resource: str,
+        resource: Union[str, List[str]],
         conditions: Optional[List[str]] = None,
         condition_file: Optional[str] = None,
         strict_conditions: bool = False,
@@ -124,6 +127,9 @@ def _generate_query_conditions(
         if conditions is None:
             conditions = []
 
+        if isinstance(resource, str):
+            resource = [resource]
+
         output = self.generate_evaluation_logic_checks(source, resource)
 
         s, a, r = z3.Strings("s a r")
@@ -134,7 +140,7 @@ def _generate_query_conditions(
         logger.debug(f"Adding constraint action is {action}")
         logger.debug(f"Adding constraint resource is {resource}")
         output.append(parse_string(a, action, wildcard=False))
-        output.append(parse_string(r, resource, wildcard=False))
+        output.append(z3.Or(*[parse_string(r, x, wildcard=False) for x in resource]))
 
         provided_conditions = set()
 
@@ -224,14 +230,57 @@ def who_can(
                 model_conditions=model_conditions,
             )
 
+            logger.debug("Adding generated query conditions")
+            # solver.set(threads=4)
             solver.add(*query_conditions)
             sat = solver.check() == z3.sat
             sources = []
             while sat:
                 s = z3.String("s")
                 m = solver.model()
                 source = m[s]
+                logger.debug(f"Found {source} as a potential candidate")
                 sources.append(str(source)[1:-1])
                 solver.add(s != source)
                 sat = solver.check() == z3.sat
             return sources
+
+    def who_can_batch_resource(
+        self,
+        action: str,
+        resources: List[str],
+        conditions: List[str] = [],
+        condition_file: Optional[str] = None,
+        strict_conditions: bool = False,
+    ) -> List[Tuple[str, str]]:
+        with self as solver:
+            logger.debug("Identifying model conditions")
+            model_conditions = get_conditions(self.model_vars)
+            logger.debug(f"Model conditions identified as: {model_conditions}")
+
+            query_conditions = self._generate_query_conditions(
+                source=None,
+                action=action,
+                resource=resources,
+                conditions=conditions,
+                condition_file=condition_file,
+                strict_conditions=strict_conditions,
+                model_conditions=model_conditions,
+            )
+
+            logger.debug("Adding generated query conditions")
+            solver.set(threads=4)
+            solver.add(*query_conditions)
+            sat = solver.check() == z3.sat
+            results = []
+            while sat:
+                s = z3.String("s")
+                r = z3.String("r")
+                m = solver.model()
+                source = m[s]
+                resource = m[r]
+                logger.debug(f"Found {source} as a potential candidate for {resource}")
+                results.append((str(source)[1:-1], str(resource)[1:-1]))
+                solver.add(z3.Not(z3.And(s == source, r == resource)))
+                sat = solver.check() == z3.sat
+            return results

iamspy/parse.py

@@ -415,38 +415,60 @@ def generate_model(data: AuthorizationDetails):
     return model
 
 
-def generate_evaluation_logic_checks(model_vars, source: Optional[str], resource: str):
-    logger.info(f"Generating evaluation logic checks for {source} against {resource}")
+def generate_evaluation_logic_checks(model_vars, source: Optional[str], resources: List[str]):
+    logger.info(f"Generating evaluation logic checks for {source} against {resources}")
     constraints = []
 
     s_account = z3.String("s_account")
     s = z3.String("s")
+    r = z3.String("r")
     constraints.append(s_account == z3.SubString(s, 13, 12))
-    resource_account = resource.split(":")[4]
-    if resource_account:
-        constraints.append(z3.String("r_account") == z3.StringVal(resource_account))
-    else:
-        constraints.append(z3.String("r_account") == z3.String(f"resource_{resource}_account"))
+    for resource in resources:
+        resource_account = resource.split(":")[4]
+        if resource_account:
+            constraints.append(
+                z3.Or(
+                    z3.Not(parse_string(r, resource, wildcard=False)),
+                    z3.String("r_account") == z3.StringVal(resource_account),
+                )
+            )
+        else:
+            constraints.append(
+                z3.Or(
+                    z3.Not(parse_string(r, resource, wildcard=False)),
+                    z3.String("r_account") == z3.String(f"resource_{resource}_account"),
+                )
+            )
     # SCPs
 
     # Resource Policy
-    resource_identifier = f"resource_{resource}"
-    resource_check = z3.Bool(resource_identifier)
-    constraints.append(z3.Bool("resource") == resource_check)
-    constraints.append(z3.Bool(f"deny_resource_{resource}") == True)  # noqa: E712
-    if resource.startswith("arn:aws:s3:::") and "/" in resource:
-        bucket_resource = resource.split("/")[0]
-        logger.info(f"Associating {bucket_resource} policy with bucket object {resource}")
-        constraints.append(z3.Bool(f"resource_{resource}") == z3.Bool(f"resource_{bucket_resource}"))
-        constraints.append(z3.Bool(f"allow_resource_{resource}") == z3.Bool(f"allow_resource_{bucket_resource}"))
-        constraints.append(z3.Bool(f"deny_resource_{resource}") == z3.Bool(f"deny_resource_{bucket_resource}"))
+    resource_check = z3.Bool("resource")
+    for resource in resources:
+        resource_identifier = f"resource_{resource}"
+        resource_specific_check = z3.Bool(resource_identifier)
         constraints.append(
-            z3.String(f"resource_{resource}_account") == z3.String(f"resource_{bucket_resource}_account")
+            z3.Or(
+                z3.Not(parse_string(r, resource, wildcard=False)),
+                resource_check == resource_specific_check,
+            )
         )
-        resource_identifier = f"resource_{bucket_resource}"
-    if resource_identifier not in model_vars:
-        logger.debug(f"Missing resource policy for {resource_identifier}, defaulting to False")
-        constraints.append(resource_check == False)  # noqa: E712
+        # TODO: Figure this out
+        constraints.append(z3.Bool(f"deny_resource_{resource}") == True)  # noqa: E712
+        if resource.startswith("arn:aws:s3:::") and "/" in resource:
+            bucket_resource = resource.split("/")[0]
+            logger.info(f"Associating {bucket_resource} policy with bucket object {resource}")
+            constraints.append(z3.Bool(f"resource_{resource}") == z3.Bool(f"resource_{bucket_resource}"))
+            constraints.append(z3.Bool(f"allow_resource_{resource}") == z3.Bool(f"allow_resource_{bucket_resource}"))
+            constraints.append(z3.Bool(f"deny_resource_{resource}") == z3.Bool(f"deny_resource_{bucket_resource}"))
+            constraints.append(
+                z3.String(f"resource_{resource}_account") == z3.String(f"resource_{bucket_resource}_account")
+            )
+            resource_identifier = f"resource_{bucket_resource}"
+        if resource_identifier not in model_vars:
+            logger.debug(f"Missing resource policy for {resource_identifier}, defaulting to False")
+            constraints.append(resource_specific_check == False)  # noqa: E712
+
+    constraints.append(z3.Or(*[parse_string(r, x, wildcard=False) for x in resources]))
 
     # Identity Policy
     identity_identifier = f"identity_{source}"
@@ -464,15 +486,18 @@ def generate_evaluation_logic_checks(model_vars, source: Optional[str], resource
             if len(x.split(":")) > 4 and (x.split(":")[5].startswith("user") or x.split(":")[5].startswith("role"))
         ]
         identity_identifiers = [
-            z3.And(
+            z3.Bool(f"test_identity_{x}")
+            == z3.And(
                 z3.Bool(x),
                 z3.Bool(f"deny_{x}"),
                 s == x.lstrip("identity_"),
                 z3.String("s_account") == z3.StringVal(x.split(":")[4]),
             )
             for x in identities
         ]
-        identity_check = z3.Or(*identity_identifiers)
+        # identity_check = z3.Or(*identity_identifiers)
+        constraints.extend(identity_identifiers)
+        identity_check = z3.Or(*[z3.Bool(f"test_identity_{x}") for x in identities])
         # TODO: This is a temporary fix for whocan, at some point need to expand this to do automatic wildcard resolution
         # for accounts external to known
         constraints.append(

pyproject.toml

@@ -21,6 +21,9 @@ pdbpp = "^0.10.3"
 [tool.poetry.scripts]
 iamspy = 'iamspy.cli:app'
 
+[tool.poetry.group.dev.dependencies]
+setuptools = "^70.0.0"
+
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"