Repository: Netflix/consoleme
File: consoleme/handlers/v2/policies.py (Python)

Code before:


class CheckPoliciesHandler(BaseAPIV2Handler):
    async def post(self):
        """
        POST /api/v2/policies/check
        """
        policy = tornado.escape.json_decode(self.request.body)
        analyzed_policy = analyze_policy_string(policy)
        findings = analyzed_policy.findings

        enhanced_findings = []

        for finding in findings:
            enhanced_finding = enhance_finding(finding)
            enhanced_findings.append(
                {
                    "issue": enhanced_finding.issue,
                    "detail": json.dumps(enhanced_finding.detail),
                    "location": enhanced_finding.location,
                    "severity": enhanced_finding.severity,
                    "title": enhanced_finding.title,
                    "description": enhanced_finding.description,
                }
            )
        try:
            client = boto3.client("accessanalyzer", region_name=config.region)
            access_analyzer_response = client.validate_policy(
                policyDocument=policy,
                policyType="IDENTITY_POLICY",  # ConsoleMe only supports identity policy analysis currently
            )
            for finding in access_analyzer_response.get("findings", []):
                for location in finding.get("locations", []):
                    enhanced_findings.append(
                        {
                            "issue": finding.get("issueCode"),
                            "detail": None,
                            "location": {
                                "line": location.get("span", {})
                                .get("start", {})
                                .get("line"),
                                "column": location.get("span", {})
                                .get("start", {})
                                .get("column"),
                                "filepath": None,
                            },
                            "severity": finding.get("findingType"),
                            "title": finding.get("issueCode"),
                            "description": finding.get("findingDetails"),
                        }
                    )

Review comment (refactor):
Let's pull this out into a utility function so we can keep this handler clean. We could even do the same with the Parliament call.

I'm thinking something like:
```python
def access_analyzer_validate(policy):
    ...
    return transformed_aa_response_dict
```

Code after:

class CheckPoliciesHandler(BaseAPIV2Handler):
    async def post(self):
        """
        POST /api/v2/policies/check
        """
        policy = tornado.escape.json_decode(self.request.body)
        if isinstance(policy, dict):
            policy = json.dumps(policy)
        log_data = {
            "function": f"{__name__}.{self.__class__.__name__}.{sys._getframe().f_code.co_name}",
            "user": self.user,
            "user-agent": self.request.headers.get("User-Agent"),
            "request_id": self.request_uuid,
            "policy": policy,
        }
        findings = await validate_iam_policy(policy, log_data)
        self.write(json.dumps(findings))
        return


class ManagedPoliciesHandler(BaseHandler):
    async def get(self, account_id):
        """
        Retrieve a list of managed policies for an account.
        """
        if config.get("policy_editor.disallow_contractors", True) and self.contractor:
            if self.user not in config.get(
                "groups.can_bypass_contractor_restrictions", []
            ):
                raise MustBeFte("Only FTEs are authorized to view this page.")
        all_account_managed_policies = await get_all_iam_managed_policies_for_account(
            account_id
        )
        self.write(json.dumps(all_account_managed_policies))