Merge branch 'release/v1.0.1'

CHANGELOG.md

@@ -4,7 +4,9 @@
 
 - Script available in pip (pip install keycloak-scanner)
 - Fix error on verbose mode
-
+- Support for scan dependencies
+- Return code 4 when vuln, 8 when no vuln but errors
+- Fix bugs
 
 ## version 0.2.0
 

keycloak_scanner/constants.py

@@ -1,7 +1,3 @@
-DEFAULT_REALMS = [
-    'master'
-]
-
 DEFAULT_CLIENTS = [
     'account',
     'admin-cli',

keycloak_scanner/logging/printlogger.py

@@ -9,6 +9,10 @@ def __init__(self, verbose=False, **kwargs):
         self.verbose = verbose
         super().__init__(**kwargs)
 
+    def warn(self, message: str):
+        print(f'[WARN] {message}')
+        super().warn(message)
+
     def info(self, message: str):
         print('[INFO] {}'.format(message))
         super().info(message)
@@ -17,3 +21,6 @@ def verbose(self, message: str, color='grey'):
         if self.verbose:
             print(colored('[VERBOSE] {}'.format(message), color))
         super().verbose(message, color)
+
+    def find(self, scanner: str, message: str, color='red'):
+        super().find(scanner, message, color)

keycloak_scanner/logging/root_logger.py

@@ -8,3 +8,8 @@ def info(self, message: str):
     def verbose(self, message: str, color: str):
         assert not hasattr(super(), 'verbose')
 
+    def find(self, scanner: str, message: str, color:str):
+        assert not hasattr(super(), 'find')
+
+    def warn(self, message: str):
+        assert not hasattr(super(), 'warn')

keycloak_scanner/logging/vuln_flag.py

@@ -0,0 +1,21 @@
+
+
+class VoidFlag:
+
+    def __init__(self, **kwargs):
+        self.has_vuln = False
+        super().__init__(**kwargs)
+
+    def set_vuln(self):
+        assert not hasattr(super(), 'set_vuln')
+
+
+class VulnFlag(VoidFlag):
+
+    def __init__(self, has_vuln=False, **kwargs):
+        self.has_vuln = has_vuln
+        super().__init__(**kwargs)
+
+    def set_vuln(self):
+        self.has_vuln = True
+        super().set_vuln()

keycloak_scanner/main.py

@@ -5,7 +5,7 @@
 import requests
 import urllib3
 
-from keycloak_scanner import custom_logging
+
 from keycloak_scanner.scanners.clients_scanner import ClientScanner
 from keycloak_scanner.scanners.form_post_xss_scanner import FormPostXssScanner
 from keycloak_scanner.scanners.none_sign_scanner import NoneSignScanner
@@ -65,8 +65,6 @@ def start(args, session: requests.Session):
     realms = args.realms.split(',') if args.realms else []
     clients = args.clients.split(',') if args.clients else []
 
-    custom_logging.verbose_mode = args.verbose
-
     if args.proxy:
         session = {'http': args.proxy, 'https': args.proxy}
 
@@ -82,12 +80,14 @@ def start(args, session: requests.Session):
         OpenRedirectScanner(base_url=args.base_url, session=session),
         FormPostXssScanner(base_url=args.base_url, session=session),
         NoneSignScanner(base_url=args.base_url, session=session)
-    ])
-    scanner.start()
+    ], verbose=args.verbose)
+    status = scanner.start()
 
-    if args.verbose:
-        print(json.dumps(scanner.scan_properties, sort_keys=True, indent=4))
-    if not args.no_fail and custom_logging.has_vuln:
+    if not args.no_fail and status.has_vulns:
         print('Fail with exit code 4 because vulnerabilities are discovered')
         sys.exit(4)
 
+    if status.has_error:
+        print('No vulns but error(s) are returned, exit with code 8')
+        sys.exit(8)
+

keycloak_scanner/masterscanner.py

@@ -1,14 +1,90 @@
-from typing import List
+import re
+from typing import List, Dict, Any, Sized
 
+from keycloak_scanner.logging.printlogger import PrintLogger
+from keycloak_scanner.logging.vuln_flag import VulnFlag
 from keycloak_scanner.scanners.scanner import Scanner
 
 
-class MasterScanner:
+def to_camel_case(text: str):
+    return re.sub('([a-z]+)([A-Z])', r'\1_\2', text).lower()
 
-    def __init__(self, scans: List[Scanner]):
+
+class DuplicateResultException(Exception):
+    pass
+
+
+class ScanResults(PrintLogger):
+
+    def __init__(self, previous_deps: Dict[str, Any], **kwargs):
+        if previous_deps is None:
+            previous_deps = {}
+        self.results: Dict[str, Any] = previous_deps
+        super().__init__(**kwargs)
+
+    def add(self, result: Any):
+        key = to_camel_case(result.__class__.__name__)
+        if key in self.results:
+            raise DuplicateResultException(result)
+        super().verbose(f'new result with key: {key} ({result})')
+        self.results[key] = result
+
+    def __repr__(self):
+        return repr(self.results)
+
+
+class NoneResultException(Exception):
+    pass
+
+
+class ScanStatus:
+
+    def __init__(self, has_error=False, has_vulns=False):
+        self.has_error = has_error
+        self.has_vulns = has_vulns
+
+
+class MasterScanner(PrintLogger):
+
+    def __init__(self, scans: List[Scanner], previous_deps: Dict[str, Any] = None, verbose=False, **kwargs):
+        if previous_deps is None:
+            previous_deps = {}
         self.scans = scans
-        self.scan_properties = {}
+        self.results = ScanResults(previous_deps, verbose=verbose)
+        super().__init__(verbose=verbose, **kwargs)
+
+    def start(self) -> ScanStatus:
+
+        has_errors = False
+        vf = VulnFlag()
+
+        for scanner in self.scans:
+
+            super().info(f'Start scanner {scanner.name()}...')
+
+            try:
+
+                result, has_vuln = scanner.perform(**self.results.results)
+
+                if has_vuln.has_vuln:
+                    vf.set_vuln()
+
+                if result is None:
+                    super().warn(f'None result for scanner {scanner.name()}')
+                    raise NoneResultException()
+
+                if isinstance(result, Sized) and len(result) == 0:
+                    super().warn(f'Result of {scanner.name()} as no results (void list), subsequent scans can be void too.')
+
+                self.results.add(result)
+
+            except TypeError as e:
+                print(f'Missing dependency for {scanner.__class__.__name__}: ({str(e)}). '
+                      f'A required previous scanner as fail.')
+                has_errors = True
+
+            except Exception as e:
+                print(f'Failed scan : {scanner.__class__.__name__}: ({str(e)}). ')
+                has_errors = True
 
-    def start(self):
-        for scan in self.scans:
-            scan.perform(scan_properties=self.scan_properties)
+        return ScanStatus(has_errors, vf.has_vuln)

keycloak_scanner/scanners/clients_scanner.py

@@ -1,36 +1,71 @@
 from typing import List
 
-from keycloak_scanner.properties import add_list
+from keycloak_scanner.logging.vuln_flag import VulnFlag
+from keycloak_scanner.scanners.realm_scanner import Realms
 from keycloak_scanner.scanners.scanner import Scanner
+from keycloak_scanner.scanners.scanner_pieces import Need2
+from keycloak_scanner.scanners.well_known_scanner import WellKnownDict
 
 URL_PATTERN = '{}/auth/realms/{}/{}'
 
 
-class ClientScanner(Scanner):
+class Client:
+
+    def __init__(self, name: str, url: str, auth_endpoint: str = None):
+        self.name = name
+        self.url = url
+        self.auth_endpoint = auth_endpoint
+
+    def __repr__(self):
+        return f"Client('{self.name}', '{self.url}', '{self.auth_endpoint}')"
+
+    def __eq__(self, other):
+        if isinstance(other, Client):
+            return self.name == other.name and self.url == other.url and self.auth_endpoint == other.auth_endpoint
+        return NotImplemented
+
+
+class Clients(List[Client]):
+    pass
+
+
+class ClientScanner(Need2[Realms, WellKnownDict], Scanner[Clients]):
 
     def __init__(self, clients: List[str], **kwargs):
         self.clients = clients
         super().__init__(**kwargs)
 
-    def perform(self, scan_properties):
-
-        realms = scan_properties['realms'].keys()
+    def perform(self, realms: Realms, well_known_dict: WellKnownDict, **kwargs) -> (Clients, VulnFlag):
 
-        scan_properties['clients'] = {}
+        result: Clients = Clients()
 
         for realm in realms:
-            for client in self.clients:
-                url = URL_PATTERN.format(super().base_url(), realm, client)
-                r = super().session().get(url)
+            for client_name in self.clients:
+                url = URL_PATTERN.format(super().base_url(), realm.name, client_name)
 
-                if r.status_code != 200:
-                    url = scan_properties['wellknowns'][realm]['authorization_endpoint']
-                    r = super().session().get(url, params={'client_id': client}, allow_redirects=False)
+                try:
+                    r = super().session().get(url)
+                    r.raise_for_status()
+
+                except Exception as e:
+                    super().info('f [ClientScanner]: {e}')
+                    url = None
+
+                try:
+
+                    auth_url = well_known_dict[realm.name].json['authorization_endpoint']
+
+                    r = super().session().get(auth_url, params={'client_id': client_name}, allow_redirects=False)
                     if r.status_code == 302:
-                        super().info('Find a client for realm {}: {}'.format(realm, client))
-                        add_list(scan_properties['clients'], realm, client)
+                        super().info('Find a client for realm {}: {}'.format(realm.name, client_name))
+                        result.append(Client(name=client_name, url=url, auth_endpoint=auth_url))
                     else:
-                        super().verbose('client {} seems to not exists'.format(client))
-                else:
-                    super().info('Find a client for realm {}: {} ({})'.format(realm, client, url))
-                    add_list(scan_properties['clients'], realm, client)
+                        super().verbose('client {} seems to not exists'.format(client_name))
+
+                except KeyError as e:
+                    print(f'realm {realm.name}\'s wellknown doesn\t existsor do not have "authorization_endpoint". ({well_known_dict})')
+                    auth_url = None
+
+                result.append(Client(name=client_name, auth_endpoint=auth_url, url=url))
+
+        return result, VulnFlag(False)

keycloak_scanner/scanners/form_post_xss_scanner.py

@@ -1,23 +1,54 @@
+from typing import Dict
+
+from keycloak_scanner.logging.vuln_flag import VulnFlag
+from keycloak_scanner.scanners.clients_scanner import Clients
+from keycloak_scanner.scanners.realm_scanner import Realms, Realm
 from keycloak_scanner.scanners.scanner import Scanner
-from keycloak_scanner.custom_logging import find
+from keycloak_scanner.scanners.scanner_pieces import Need3
+from keycloak_scanner.scanners.well_known_scanner import WellKnownDict
+
+
+class FormPostXssResult:
+
+    def __init__(self, realm: Realm, is_vulnerable: bool):
+        self.realm = realm
+        self.is_vulnerable = is_vulnerable
+
+    def __repr__(self):
+        return f'FormPostXssResult({repr(self.realm)}, {self.is_vulnerable})'
+
+    def __eq__(self, other):
+        if isinstance(other, FormPostXssResult):
+            return self.realm == other.realm and self.is_vulnerable == other.is_vulnerable
+        return NotImplemented
+
+
+class FormPostXssResults(Dict[str, FormPostXssResult]):
+    pass
 
 
-class FormPostXssScanner(Scanner):
+class FormPostXssScanner(Need3[Realms, Clients, WellKnownDict], Scanner[FormPostXssResults]):
 
     def __init__(self, **kwars):
         super().__init__(**kwars)
 
-    def perform(self, scan_properties):
+    def perform(self, realms: Realms, clients: Clients, well_known_dict: WellKnownDict, **kwargs) -> (FormPostXssResults, VulnFlag):
 
-        realms = scan_properties['realms'].keys()
+        results = FormPostXssResults()
+
+        vf = VulnFlag()
 
         for realm in realms:
-            clients = scan_properties['clients'][realm]
-            well_known = scan_properties['wellknowns'][realm]
-            if 'form_post' not in well_known['response_modes_supported']:
-                super().verbose('post_form not in supported response types, can\' test CVE-2018-14655 for realm {}'.format(realm))
+
+            well_known = well_known_dict[realm.name]
+
+            vulnerable = False
+
+            if 'form_post' not in well_known.json['response_modes_supported']:
+                super().verbose(f'post_form not in supported response types, can\' test CVE-2018-14655 for realm {realm}')
+
             else:
-                url = well_known['authorization_endpoint']
+                url = well_known.json['authorization_endpoint']
 
                 for client in clients:
 
@@ -32,4 +63,10 @@ def perform(self, scan_properties):
 
                     if r.status_code == 200:
                         if payload in r.text:
-                            find('XSS-CVE2018-14655', 'Vulnerable to CVE 2018 14655 realm:{}, client:{}'.format(realm, client))
+                            super().find('XSS-CVE2018-14655', 'Vulnerable to CVE 2018 14655 realm:{}, client:{}'.format(realm, client))
+                            vulnerable = True
+                            vf.set_vuln()
+
+            results[realm.name] = FormPostXssResult(realm, vulnerable)
+
+        return results, vf