Merge pull request #66 from atlassian-labs/authnchecks

Authnchecks

.github/workflows/csrt-lint-and-test.yml

@@ -30,9 +30,9 @@ jobs:
     - name: Lint with flake8
       run: |
         pipenv run lint
-    - name: Check dependencies for security issues
-      run: |
-        pipenv check
+#    - name: Check dependencies for security issues
+#      run: |
+#        pipenv check
     - name: Test with pytest (retry up to 3 times)
       run: |
         pipenv run test || pipenv run test || pipenv run test

Pipfile

@@ -20,6 +20,7 @@ tldextract = "*"
 sslyze = "*"
 dnspython = "*"
 python-json-logger = "*"
+cryptography = "*"
 
 [requires]
 python_version = "3.9"

README.md

@@ -19,7 +19,7 @@ Common usage:
 
 CSRT with all arguments:
 
-`pipenv run python main.py url-to-atlassian-connect-json --debug=True/False --out_dir=./out --skip_branding=True/False --timeout=30 --json_logging=True/False`
+`pipenv run python main.py url-to-atlassian-connect-json --debug=True/False --out_dir=./out --skip_branding=True/False --timeout=30 --json_logging=True/False --user_jwt=<jwt_token> --authz_only=True/False`
 
 ### Docker Usage
 
@@ -31,14 +31,15 @@ Run the following from the project root:
 2. `docker run -v $(pwd)/out:/app/out connect-security-req-tester <url of descriptor>`
 
 ### Arguments
-| Argument | Argument Description |
-|----------|----------------------|
-|--timeout          | Defines how long CSRT will wait on web requests before timing out, **default: 30 seconds**        |
-|--skip_branding    | Whether or not to skip branding checks, **default: False**                                        |
-|--out_dir          | The output directory where results are stored, **default: ./out**                                 |
-|--json_logging     | Whether or not to log output in a JSON format, **default: False**                                 |
-|--debug            | Sets logging to DEBUG for more verbose logging, **default: False**                                |
-
+| Argument | Argument Description                                                                       |
+|----------|--------------------------------------------------------------------------------------------|
+|--timeout          | Defines how long CSRT will wait on web requests before timing out, **default: 30 seconds** |
+|--skip_branding    | Whether or not to skip branding checks, **default: True**                                  |
+|--out_dir          | The output directory where results are stored, **default: ./out**                          |
+|--json_logging     | Whether or not to log output in a JSON format, **default: False**                          |
+|--debug            | Sets logging to DEBUG for more verbose logging, **default: False**                         |
+|--user_jwt         | A **user** JWT token to use for authorization check on admin endpoints, **default: None**  |
+|--authz_only       | Only run and report authorization check, **default: False**                                |
 ### Environment Variables
 | Variable | Description |
 |----------|-------------|
@@ -51,7 +52,13 @@ This tool assumes your connect app is reachable by the machine running this tool
 
 This tool will make network requests on from your computer. Please ensure this is allowed from your organization if running this from a monitored network.
 
-**Tip**: Use a proxy by setting `OUTBOUND_PROXY` to your organization's proxy server if your app needs to be accessed via a proxy server.
+**Authorization Check**:
+* This tool also runs authorization check on admin endpoints to report any authorization bypass issues. If your app uses admin modules, and they need to be authenticated to access admin endpoints, you can pass a user JWT token via the `--user_jwt` argument. This will allow the tool to make requests to admin endpoints using user authentication information and test for authorization bypass issues. If you do not pass a user JWT token, the tool will skip authorization checks on admin endpoints.
+* You can generate a user JWT token for testing by following the instructions at: [https://developer.atlassian.com/cloud/jira/platform/understanding-jwt/](https://developer.atlassian.com/cloud/jira/platform/understanding-jwt/) and use a shared secret received on your test instance for signing or capture a context token by entering `AP.context.getToken(console.log)` in the browser’s dev console when you load the app in Jira/Confluence.
+* Additionally, if you only want to run Authorization check and not the entire suite of checks in this tool, you can pass the `--authz_only` argument.
+
+**Tips**: 
+* Use a proxy by setting `OUTBOUND_PROXY` to your organization's proxy server if your app needs to be accessed via a proxy server.
 
 Additional information about the Atlassian Connect Security Requirements can be found at: [https://developer.atlassian.com/platform/marketplace/security-requirements-more-info/](https://developer.atlassian.com/platform/marketplace/security-requirements-more-info/)
 

analyzers/descriptor_analyzer.py

@@ -5,8 +5,9 @@
 from models.descriptor_result import DescriptorResult
 from models.requirements import Requirements, RequirementsResult
 from reports.constants import (MISSING_ATTRS_SESSION_COOKIE,
-                               MISSING_AUTHN_AUTHZ, MISSING_CACHE_HEADERS,
-                               MISSING_REF_HEADERS, NO_ISSUES, NO_AUTH_PROOF, VALID_AUTH_PROOF, REQ_TITLES)
+                               MISSING_AUTHN, MISSING_CACHE_HEADERS,
+                               MISSING_REF_HEADERS, NO_ISSUES, NO_AUTH_PROOF, VALID_AUTH_PROOF, VALID_AUTHZ_PROOF, REQ_TITLES,
+                               MISSING_SIGNED_INSTALL_AUTHN, MISSING_AUTHZ)
 
 REQ_CACHE_HEADERS = ['no-cache', 'no-store']
 REF_DENYLIST = ['no-referrer-when-downgrade', 'unsafe-url']
@@ -66,9 +67,13 @@ def _check_cookie_headers(self) -> Tuple[bool, List[str]]:
 
         return passed, proof
 
-    def _check_authn_authz(self) -> Tuple[bool, List[str]]:
+    def _check_authn_authz(self) -> Tuple[bool, List[str], bool, List[str], bool, List[str]]:
         passed = True
         proof: List[str] = []
+        signed_install_passed = True
+        signed_install_proof: List[str] = []
+        authz_passed = True
+        authz_proof: List[str] = []
         scan_res = self.scan.scan_results
 
         # Don't check authentication if the app doesn't have an authentication method.
@@ -77,61 +82,107 @@ def _check_authn_authz(self) -> Tuple[bool, List[str]]:
         use_authentication = (False if authentication_method is None else authentication_method.get("type") == "jwt")
         if not use_authentication:
             proof.append(NO_AUTH_PROOF)
-            return passed, proof
+            return passed, proof, signed_install_passed, signed_install_proof, authz_passed, authz_proof
 
         for link in scan_res:
-            res_code = int(scan_res[link].res_code)
+            res_code = int(scan_res[link].res_code) if scan_res[link].res_code else 0
             auth_header = scan_res[link].auth_header
             req_method = scan_res[link].req_method
+            response = scan_res[link].response
+            authz_req_method = scan_res[link].authz_req_method
+            authz_code = int(scan_res[link].authz_code) if scan_res[link].authz_code else 0
+            authz_header = scan_res[link].authz_header
+
+            # Check for invalid responses in the body before failing the authn check
+            invalid_responses = ['Invalid JWT', 'unauthorized', 'forbidden', 'error', 'unlicensed', 'not licensed',
+                                 'no license', 'invalid', '401', '403', '404', '500']
+            invalid_response = False
+            if any(str(x).lower() in str(response).lower() for x in invalid_responses):
+                invalid_response = True
 
             # We shouldn't be able to visit this link if the app uses authentication.
-            if res_code >= 200 and res_code < 400:
-                passed = False
-                proof_text = f"{link} | Res Code: {res_code} Req Method: {req_method} Auth Header: {auth_header}"
-                proof.append(proof_text)
+            if res_code >= 200 and res_code < 400 and not invalid_response:
+                if any(x in link for x in ('installed', 'uninstalled')):
+                    signed_install_passed = False
+                    signed_install_proof_text = f"Lifecycle endpoint: {link} | Res Code: {res_code}" \
+                                                f" Auth Header: {auth_header}"
+                    signed_install_proof.append(signed_install_proof_text)
+
+                else:
+                    passed = False
+                    proof_text = f"{link} | Res Code: {res_code} Req Method: {req_method} Auth Header: {auth_header}"
+                    proof.append(proof_text)
+
+            # similarly check for authorization status codes for authorization bypass
+            if authz_code >= 200 and authz_code < 400:
+                authz_passed = False
+                authz_proof_text = (f"{link} | Authz Res Code: {authz_code} Req Method: {authz_req_method}"
+                                    f" Authz Header: {authz_header}")
+                authz_proof.append(authz_proof_text)
 
         if passed:
             proof.append(VALID_AUTH_PROOF)
+        if authz_passed:
+            authz_proof.append(VALID_AUTHZ_PROOF)
 
-        return passed, proof
+        return passed, proof, signed_install_passed, signed_install_proof, authz_passed, authz_proof
 
-    def analyze(self) -> Requirements:
+    def analyze(self, authz_only=False) -> Requirements:
         cache_passed, cache_proof = self._check_cache_headers()
         ref_passed, ref_proof = self._check_referrer_headers()
         cookies_passed, cookies_proof = self._check_cookie_headers()
-        auth_passed, auth_proof = self._check_authn_authz()
-
-        req2 = RequirementsResult(
-            passed=cache_passed,
-            description=[NO_ISSUES] if cache_passed else [MISSING_CACHE_HEADERS],
-            proof=cache_proof,
-            title=REQ_TITLES['2']
-        )
+        (auth_passed, auth_proof, signed_install_passed, signed_install_proof,
+         authz_passed, authz_proof) = self._check_authn_authz()
 
-        req5 = RequirementsResult(
+        req1_1 = RequirementsResult(
             passed=auth_passed,
-            description=[NO_ISSUES] if auth_passed else [MISSING_AUTHN_AUTHZ],
+            description=[NO_ISSUES] if auth_passed else [MISSING_AUTHN],
             proof=auth_proof,
-            title=REQ_TITLES['5']
+            title=REQ_TITLES['1.1']
         )
 
-        req11 = RequirementsResult(
-            passed=cookies_passed,
-            description=[NO_ISSUES] if cookies_passed else [MISSING_ATTRS_SESSION_COOKIE],
-            proof=cookies_proof,
-            title=REQ_TITLES['11']
+        req1_2 = RequirementsResult(
+            passed=authz_passed,
+            description=[NO_ISSUES] if authz_passed else [MISSING_AUTHZ],
+            proof=authz_proof,
+            title=REQ_TITLES['1.2']
+        )
+
+        req1_4 = RequirementsResult(
+            passed=signed_install_passed,
+            description=[NO_ISSUES] if signed_install_passed else [MISSING_SIGNED_INSTALL_AUTHN],
+            proof=signed_install_proof,
+            title=REQ_TITLES['1.4']
         )
 
-        req12 = RequirementsResult(
+        req7_2 = RequirementsResult(
             passed=ref_passed,
             description=[NO_ISSUES] if ref_passed else [MISSING_REF_HEADERS],
             proof=ref_proof,
-            title=REQ_TITLES['12']
+            title=REQ_TITLES['7.2']
+        )
+
+        req7_3 = RequirementsResult(
+            passed=cache_passed,
+            description=[NO_ISSUES] if cache_passed else [MISSING_CACHE_HEADERS],
+            proof=cache_proof,
+            title=REQ_TITLES['7.3']
+        )
+
+        req7_4 = RequirementsResult(
+            passed=cookies_passed,
+            description=[NO_ISSUES] if cookies_passed else [MISSING_ATTRS_SESSION_COOKIE],
+            proof=cookies_proof,
+            title=REQ_TITLES['7.4']
         )
 
-        self.reqs.req2 = req2
-        self.reqs.req5 = req5
-        self.reqs.req11 = req11
-        self.reqs.req12 = req12
+        # Skip reporting other checks if we only run authz check
+        if not authz_only:
+            self.reqs.req1_1 = req1_1
+            self.reqs.req1_4 = req1_4
+            self.reqs.req7_2 = req7_2
+            self.reqs.req7_3 = req7_3
+            self.reqs.req7_4 = req7_4
+        self.reqs.req1_2 = req1_2
 
         return self.reqs

analyzers/hsts_analyzer.py

@@ -21,13 +21,13 @@ def _check_header_present(self) -> tuple[bool, list[str]]:
     def analyze(self) -> Requirements:
         header_passed, header_proof = self._check_header_present()
 
-        req1_2 = RequirementsResult(
+        req3_0 = RequirementsResult(
             passed=header_passed,
             description=[NO_ISSUES] if header_passed else [HSTS_MISSING],
             proof=header_proof,
-            title=REQ_TITLES['1.2']
+            title=REQ_TITLES['3.0']
         )
 
-        self.reqs.req1_2 = req1_2
+        self.reqs.req3_0 = req3_0
 
         return self.reqs

analyzers/tls_analyzer.py

@@ -37,20 +37,20 @@ def analyze(self) -> Requirements:
         tls_passed, tls_proof = self._check_tls_versions()
         cert_passed, cert_proof = self._check_cert_valid()
 
-        req1_1 = RequirementsResult(
+        req3 = RequirementsResult(
             passed=tls_passed,
             description=[NO_ISSUES] if tls_passed else [TLS_PROTOCOLS],
             proof=tls_proof,
-            title=REQ_TITLES['1.1']
+            title=REQ_TITLES['3']
         )
-        req3 = RequirementsResult(
+        req6_2 = RequirementsResult(
             passed=cert_passed,
             description=[NO_ISSUES] if cert_passed else [CERT_NOT_VALID],
             proof=cert_proof,
-            title=REQ_TITLES['3']
+            title=REQ_TITLES['6.2']
         )
 
-        self.reqs.req1_1 = req1_1
         self.reqs.req3 = req3
+        self.reqs.req6_2 = req6_2
 
         return self.reqs

main.py

@@ -18,7 +18,8 @@
 from utils.app_validator import AppValidator
 
 
-def main(descriptor_url, skip_branding=False, debug=False, timeout=30, out_dir='out', json_logging=False):
+def main(descriptor_url, skip_branding=True, debug=False, timeout=30, out_dir='out', json_logging=False,
+         user_jwt=None, authz_only=False):
     # Setup our logging
     setup_logging('connect-security-requirements-tester', debug, json_logging)
     logging.info(f"CSRT Scan started at: {(start := datetime.utcnow())}")
@@ -29,13 +30,18 @@ def main(descriptor_url, skip_branding=False, debug=False, timeout=30, out_dir='
     app_url = validator.get_test_url()
 
     # Run our scans -- TLS/HSTS/Descriptor
-    tls_scan = TlsScan(app_url)
-    hsts_scan = HstsScan(app_url, timeout)
-    descriptor_scan = DescriptorScan(descriptor_url, descriptor, timeout)
+    # Skip TLS/HSTS scans if we're only doing authorization checks
+    if not authz_only:
+        tls_scan = TlsScan(app_url)
+        hsts_scan = HstsScan(app_url, timeout)
+        tls_res = tls_scan.scan()
+        hsts_res = hsts_scan.scan()
+    else:
+        tls_res = None
+        hsts_res = None
 
-    tls_res = tls_scan.scan()
-    hsts_res = hsts_scan.scan()
-    descriptor_res = descriptor_scan.scan()
+    descriptor_scan = DescriptorScan(descriptor_url, descriptor, timeout)
+    descriptor_res = descriptor_scan.scan(user_jwt)
 
     # Analyze the results from the scans
     results = Results(
@@ -44,21 +50,23 @@ def main(descriptor_url, skip_branding=False, debug=False, timeout=30, out_dir='
         base_url=descriptor_res.base_url,
         app_descriptor_url=descriptor_res.app_descriptor_url,
         requirements=Requirements(),
-        tls_scan_raw=json.dumps(tls_res.to_json(), indent=3),
+        tls_scan_raw=json.dumps(tls_res.to_json(), indent=3) if tls_res else None,
         descriptor_scan_raw=json.dumps(descriptor_res.to_json(), indent=3),
         errors=descriptor_res.link_errors
     )
 
     logging.info('Starting analysis of results...')
 
-    tls_analyzer = TlsAnalyzer(tls_res, results.requirements)
-    results.requirements = tls_analyzer.analyze()
+    # Skip analyzing TLS/HSTS scans if we're only doing authorization checks
+    if not authz_only:
+        tls_analyzer = TlsAnalyzer(tls_res, results.requirements)
+        results.requirements = tls_analyzer.analyze()
 
-    hsts_analyzer = HstsAnalyzer(hsts_res, results.requirements)
-    results.requirements = hsts_analyzer.analyze()
+        hsts_analyzer = HstsAnalyzer(hsts_res, results.requirements)
+        results.requirements = hsts_analyzer.analyze()
 
     descriptor_analyzer = DescriptorAnalyzer(descriptor_res, results.requirements)
-    results.requirements = descriptor_analyzer.analyze()
+    results.requirements = descriptor_analyzer.analyze(authz_only)
 
     if not skip_branding:
         branding_analyzer = BrandingAnalyzer(descriptor_res.links, descriptor_res.name, results.requirements)

models/descriptor_result.py

@@ -9,6 +9,10 @@ class DescriptorLink(JsonObject):
     auth_header = StringProperty()
     req_method = StringProperty()
     res_code = StringProperty()
+    response = StringProperty()
+    authz_req_method = StringProperty()
+    authz_code = StringProperty()
+    authz_header = StringProperty()
 
 
 class DescriptorResult(JsonObject):
@@ -21,3 +25,4 @@ class DescriptorResult(JsonObject):
     links = ListProperty(StringProperty())
     link_errors = DictProperty()
     scan_results = DictProperty(DescriptorLink)
+    response = StringProperty()

models/requirements.py

@@ -15,20 +15,20 @@ def was_scanned(self) -> bool:
 class Requirements(JsonObject):
     req1_1 = ObjectProperty(RequirementsResult, name='1.1')
     req1_2 = ObjectProperty(RequirementsResult, name='1.2')
+    req1_4 = ObjectProperty(RequirementsResult, name='1.4')
     req2 = ObjectProperty(RequirementsResult, name='2')
     req3 = ObjectProperty(RequirementsResult, name='3')
-    req4 = ObjectProperty(RequirementsResult, name='4')
+    req3_0 = ObjectProperty(RequirementsResult, name='3.0')
     req5 = ObjectProperty(RequirementsResult, name='5')
-    req6 = ObjectProperty(RequirementsResult, name='6')
-    req7 = ObjectProperty(RequirementsResult, name='7')
-    req8 = ObjectProperty(RequirementsResult, name='8')
+    req6_2 = ObjectProperty(RequirementsResult, name='6.2')
+    req6_3 = ObjectProperty(RequirementsResult, name='6.3')
+    req7_2 = ObjectProperty(RequirementsResult, name='7.2')
+    req7_3 = ObjectProperty(RequirementsResult, name='7.3')
+    req7_4 = ObjectProperty(RequirementsResult, name='7.4')
+    req8_1 = ObjectProperty(RequirementsResult, name='8.1')
     req9 = ObjectProperty(RequirementsResult, name='9')
     req10 = ObjectProperty(RequirementsResult, name='10')
     req11 = ObjectProperty(RequirementsResult, name='11')
-    req12 = ObjectProperty(RequirementsResult, name='12')
-    req13 = ObjectProperty(RequirementsResult, name='13')
-    req14 = ObjectProperty(RequirementsResult, name='14')
-    req15 = ObjectProperty(RequirementsResult, name='15')
     req16 = ObjectProperty(RequirementsResult, name='16')