Add SBS login monitoring and timestamp

getmonitor.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+import sys
+
+# getmonitor.py is used to retrieve the monitor values
+# using Zabbix system.run[]
+
+
+def get(env, command):
+    try:
+        with open(f'status/{env}.log', 'r') as f:
+            time = f.readline()
+            test = f.readline()
+            login = f.readline()
+            if command == 'time':
+                return time.strip()
+            elif command == 'test':
+                return test.strip()
+            elif command == 'login':
+                return login.strip()
+            elif command == 'pam':
+                return login.strip()
+    except Exception:
+        return "error\n"
+
+
+if __name__ == '__main__':
+    if len(sys.argv) < 3:
+        sys.exit(sys.argv[0] + "  <env> <command>")
+    env = sys.argv[1]
+    command = sys.argv[2]
+    print(get(env, command))

pam-monitor.py

@@ -0,0 +1,148 @@
+#!/usr/bin/env python
+import sys
+import time
+import yaml
+import requests
+import json
+import re
+import traceback
+
+from selenium.webdriver import Remote, ChromeOptions
+from selenium.webdriver.support.wait import WebDriverWait
+from selenium.webdriver.support.expected_conditions import (staleness_of, presence_of_element_located,
+                                                            title_contains)
+from selenium.webdriver.common.by import By
+
+if len(sys.argv) < 2:
+    sys.exit(sys.argv[0] + "  <argument>")
+
+print(f"= READING {sys.argv[1]} ====", file=sys.stderr)
+with open(sys.argv[1], 'r') as f:
+    try:
+        config = yaml.load(f, Loader=yaml.FullLoader)
+    except yaml.YAMLError as exc:
+        print(exc)
+
+KEY = 'pam_monitor'
+config = config[KEY]
+
+start_payload = {
+    # "user_id": "ascz",
+    "attribute": "username",
+    "cache_duration": 60,
+    "GIT_COMMIT": "monitor",
+    "JSONPARSER_GIT_COMMIT": "monitor",
+}
+regex = r"(https?://.*)\s"
+xpath_login_button = "//span[@class='textual' and text()='Log in']"
+xpath_code_element = "//span[@class='value']"
+
+
+print("= Starting Chrome ===", file=sys.stderr)
+options = ChromeOptions()
+options.add_argument('--headless')
+browser = Remote("http://127.0.0.1:4444", options=options)
+send_command = ('POST', '/session/$sessionId/chromium/send_command')
+browser.command_executor._commands['SEND_COMMAND'] = send_command
+browser.implicitly_wait(1)
+wait = WebDriverWait(browser, timeout=2)
+
+
+try:
+    for url, values in config.items():
+        token = values['token']
+        account = values['account']
+        (username, password) = account.split('.')
+        # print(f"{url}, {token}", file=sys.stderr)
+        # print(f"{username}:{password}", file=sys.stderr)
+        health = f'{url}health'
+        headers = {
+            "Authorization": f"Bearer {token}",
+        }
+
+        r = requests.post(f"{url}pam-weblogin/start", json=start_payload, headers=headers)
+        j = json.loads(r.text)
+
+        session_id = j['session_id']
+        challenge = j['challenge']
+        link = re.findall(regex, challenge)[0]
+        print(link, file=sys.stderr)
+        # break
+
+        # Wait for SBS health up
+        status = ""
+        while status != "UP":
+            browser.get(health)
+            state = json.loads(browser.find_element(By.XPATH, "//pre").text)
+            status = state.get("status")
+            time.sleep(1)
+
+        # Clear cookies
+        browser.execute('SEND_COMMAND', dict(cmd='Network.clearBrowserCookies', params={}))
+
+        # Get SBS link page
+        browser.get(link)
+
+        # Wait for loginpage to load
+        wait.until(presence_of_element_located((By.XPATH, xpath_login_button)),
+                   'Timeout waiting for Profile')
+
+        # Click login
+        browser.find_element(By.XPATH, xpath_login_button).click()
+
+        # Wait for discovery to load
+        wait.until(title_contains('SURF Research Access Management'),
+                   'Timeout waiting for discovery')
+
+        # Choose Monitoring IdP
+        try:
+            xpath = "//div[@class='primary' and text()='SRAM Monitoring IdP']"
+            monitor = browser.find_element(By.XPATH, xpath)
+        except Exception:
+            search = browser.find_element(By.ID, 'searchinput')
+            search.send_keys('monitoring')
+            xpath = "//div[@class='text-truncate label primary' and text()='SRAM Monitoring IdP']"
+            monitor = browser.find_element(By.XPATH, xpath)
+
+        monitor.click()
+
+        wait.until(staleness_of(monitor))
+
+        # Login as user
+        browser.find_element(By.ID, 'username').send_keys(username)
+        browser.find_element(By.ID, 'password').send_keys(password)
+        browser.find_element(By.NAME, 'f').submit()
+
+        # Wait for Verification code to load
+        wait.until(presence_of_element_located((By.XPATH, xpath_code_element)),
+                   'Timeout waiting for Profile')
+
+        # Test admin attributes
+        code = browser.find_elements(By.XPATH, xpath_code_element)[0].text
+
+        print(f"code: {code}", file=sys.stderr)
+
+        pin_payload = {
+            "session_id": session_id,
+            "pin": code,
+        }
+
+        r = requests.post(f"{url}pam-weblogin/check-pin", json=pin_payload, headers=headers)
+        j = json.loads(r.text)
+
+        result = j['result']
+
+        if not result == "SUCCESS":
+          raise Exception("FAIL")
+
+        print(f"result: {result}", file=sys.stderr)
+        print("= OK =======", file=sys.stderr)
+
+    print("OK")
+    browser.quit()
+
+except Exception as e:
+    tr = traceback.extract_tb(e.__traceback__)[0]
+    print(f"error {type(e).__name__} on line {tr.lineno} of '{tr.filename}'")
+    browser.quit()
+    exit(1)

requirements.txt

@@ -1,2 +1,3 @@
 PyYAML
 selenium
+requests

run_sram_monitoring.sh

@@ -10,18 +10,29 @@ if [ ! -f ${ENV}.yml ]; then
   exit 1
 fi
 
-docker run --rm -d --name chrome -p 4444:4444 selenium/standalone-chrome >/dev/null 2>&1
-while ! curl --output /dev/null --silent --head http://localhost:4444/wd/hub; do
+docker run --shm-size=2g --rm -d --name chrome -p 4444:4444 selenium/standalone-chrome >/dev/null 2>&1
+while ! curl --output /dev/null --silent --head http://localhost:4444/wd/hub/status; do
 #   echo -n "."
   sleep 1
 done
 # echo " Up!"
 
+LOGFILE="status/${ENV}.log"
+
+#date
+date --utc +"%s" > $LOGFILE
+
+# monitoring test
 OUTPUT=$(python3 sram_monitoring_test.py ${ENV}.yml)
-RESULT=$?
+echo $OUTPUT >> $LOGFILE
+
+# SBS login test
+OUTPUT=$(python3 sbs-login.py ${ENV}.yml)
+echo $OUTPUT >> $LOGFILE
+
+# SBS login test
+OUTPUT=$(python3 pam-monitor.py ${ENV}.yml)
+echo $OUTPUT >> $LOGFILE
 
 docker stop chrome >/dev/null 2>&1
 # echo "Down"
-
-echo $OUTPUT > status/${ENV}.log
-exit $RESULT

sbs-login.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python
+
+from selenium.webdriver import Remote, ChromeOptions
+from selenium.webdriver.support.wait import WebDriverWait
+from selenium.webdriver.support.expected_conditions import (staleness_of, presence_of_element_located,
+                                                            title_contains)
+from selenium.webdriver.common.by import By
+
+import sys
+import yaml
+import json
+import time
+import traceback
+
+if len(sys.argv) < 2:
+    sys.exit(sys.argv[0] + "  <argument>")
+
+print(f"= READING {sys.argv[1]} ====", file=sys.stderr)
+with open(sys.argv[1], 'r') as f:
+    try:
+        config = yaml.load(f, Loader=yaml.FullLoader)
+    except yaml.YAMLError as exc:
+        print(exc)
+
+KEY = 'sbs_login'
+config = config[KEY]
+
+xpath_login_button = '//button[span[text()="Log in"]]'
+xpath_logo = '//a[@class="logo"]//span[text()="Research Access Management"]'
+
+print("= Starting Chrome ===", file=sys.stderr)
+options = ChromeOptions()
+options.add_argument('--headless')
+browser = Remote("http://127.0.0.1:4444", options=options)
+send_command = ('POST', '/session/$sessionId/chromium/send_command')
+browser.command_executor._commands['SEND_COMMAND'] = send_command
+browser.implicitly_wait(1)
+wait = WebDriverWait(browser, timeout=2)
+
+
+try:
+    for (url, login) in config.items():
+        print(url, file=sys.stderr)
+        for (account, name) in login.items():
+            (username, password) = account.split('.')
+            print("============", file=sys.stderr)
+            print(username, name, file=sys.stderr)
+            # print(password, file=sys.stderr)
+            # print(name, file=sys.stderr)
+
+            health = f'{url}health'
+            start = f'{url}profile'
+
+            # Wait for SBS health up
+            status = ""
+            while status != "UP":
+                browser.get(health)
+                state = json.loads(browser.find_element(By.XPATH, "//pre").text)
+                status = state.get("status")
+                time.sleep(1)
+
+            # Clear cookies
+            browser.execute('SEND_COMMAND', dict(cmd='Network.clearBrowserCookies', params={}))
+
+            # Get SBS profile page
+            browser.get(start)
+
+            # Wait for discovery to load
+            wait.until(title_contains('SURF Research Access Management'),
+                       'Timeout waiting for discovery')
+
+            # Choose Monitoring IdP
+            try:
+                xpath = "//div[@class='primary' and text()='SRAM Monitoring IdP']"
+                monitor = browser.find_element(By.XPATH, xpath)
+            except Exception:
+                search = browser.find_element(By.ID, 'searchinput')
+                search.send_keys('monitoring')
+                xpath = "//div[@class='text-truncate label primary' and text()='SRAM Monitoring IdP']"
+                monitor = browser.find_element(By.XPATH, xpath)
+
+            monitor.click()
+
+            wait.until(staleness_of(monitor))
+
+            # Login as user
+            browser.find_element(By.ID, 'username').send_keys(username)
+            browser.find_element(By.ID, 'password').send_keys(password)
+            browser.find_element(By.NAME, 'f').submit()
+
+            # Wait for Profile to load
+            wait.until(presence_of_element_located((By.XPATH, "//h2[text()='Your profile']")),
+                       'Timeout waiting for Profile')
+
+            # Test admin attributes
+            attributes = browser.find_elements(By.XPATH, "//table[@class='my-attributes']/*/*/*")
+            # for a in attributes:
+            #     print(f"a.text: {a.text}")
+            assert (name in [a.text for a in attributes]), "No valid admin profile found"
+            print("= OK =======", file=sys.stderr)
+        print("OK")
+    browser.quit()
+    print("============", file=sys.stderr)
+    exit(0)
+
+except Exception as e:
+    # url = browser.current_url
+    # print(f"url: {url}")
+
+    tr = traceback.extract_tb(e.__traceback__)[0]
+    print(f"error {type(e).__name__} on line {tr.lineno} of '{tr.filename}'")
+    browser.quit()
+
+    print("", end="", flush=True)
+    exit(1)

sram_monitoring_test.py

@@ -23,7 +23,7 @@
 KEY = 'sram_monitoring'
 config = config[KEY]
 
-print(f"= Starting Chrome ===", file=sys.stderr)
+print("= Starting Chrome ===", file=sys.stderr)
 options = ChromeOptions()
 options.add_argument('--headless')
 browser = Remote("http://127.0.0.1:4444/wd/hub", options=options)
@@ -92,7 +92,7 @@ def test_user(start, user, userinfo):
         print(f"FAILED on user={user['name']} page={start}")
         # page = browser.page_source
         # print("page: {}".format(page))
-        # browser.close()
+        browser.quit()
 
         # exit without error, so that the next tests will still run
         exit(0)
@@ -110,5 +110,5 @@ def test_user(start, user, userinfo):
         test_user(startpage, user, userinfo)
 
 # Close browser
-# browser.close()
+browser.quit()
 print('OK')