#!/usr/bin/env bash

# CAUTION: Okta only allows password reset where the previous password is supplied.
# If the operation fails, this script will NOT roll back to the old password.
# Please ensure that the old password is correct before running this script.
# This script is provided as an example only and is not supported by Keeper Security.

# This will be executed as the following
# history -c && echo "BASE64STRING==" | /path/to/script.sh

# Without this the script might report a success
# if something fails in the script.
set -o pipefail -e

IFS= read -r params
json=$(echo "$params" | base64 -d)

# There is no built in JSON parser.
# In order to parse JSON, a tool like jq or fx is required.
$(echo "$json" | jq -r 'keys[] as $k | "export \($k)=\(.[$k])"')

# Set your Okta API token and organization URL
recordJson=$(echo "$records" | base64 -d)
OKTA_API_RECORD="Okta API Access Details"
OKTA_API_TOKEN=$(echo "$recordJson" | jq -r ".[] | select(.title == \"$OKTA_API_RECORD\").password")
OKTA_ORG_URL=$(echo "$recordJson" | jq -r ".[] | select(.title == \"$OKTA_API_RECORD\").url")

# Set the user's ID and the new password
USER_ID="$user"
OLD_PASSWORD="$oldPassword"
NEW_PASSWORD="$newPassword"

curl -v -X POST \
-H "Accept: application/json" \
-H "Content-Type: application/json" \
-H "Authorization: SSWS "$OKTA_API_TOKEN"" \
-d '{
    "oldPassword": "'"$OLD_PASSWORD"'",
    "newPassword": "'"$NEW_PASSWORD"'"
}' "$OKTA_ORG_URL/api/v1/users/$USER_ID/credentials/change_password"

#!/usr/local/bin/python3

import asyncio
import sys
import base64
import json

# CAUTION: Okta SDK only allows password reset where the previous password is supplied.
# If the operation fails, this script will NOT roll back to the old password.
# Please ensure that the old password is correct before running this script.

# Display Python version
print(f"# Python version: {sys.version}")

# Display installed packages. Uncomment lines below to inspect the environment to make sure that the
# required packages are installed.

# import pkg_resources
# print("# \n# Print installed packages for debugging purposes:")
# installed_packages = pkg_resources.working_set
# installed_packages_list = sorted(["%s==%s" % (i.key, i.version) for i in installed_packages])
# for m in installed_packages_list:
#     print(f"  {m}")

# Import Okta modules
try:
    import okta
    from okta.client import Client as OktaClient
except ImportError:
    print("# Error: Okta client package is not installed. Run 'pip install okta' to install.")
    exit(1)


# Function to find a password by its title in the records
def find_password_by_title(records, target_title):
    """Search for a password by its title in the given records."""
    for record in records:
        if record['title'] == target_title:
            return record['password']
    return None  # Return None if no matching record is found


# Initialize variables
okta_api_access_details_title = 'Okta API Access Details'
okta_api_access_details_record = None

# Read and decode input parameters from stdin
for base64_params in sys.stdin:
    params = json.loads(base64.b64decode(base64_params).decode())

    # Print available params for debugging purposes
    # print(f"# \n# Available params for the script:")
    # for key, value in params.items():
    #     print(f"#     {key}={value}")

    # Decode and load records
    records = json.loads(base64.b64decode(params.get('records')).decode())

    # Find the Okta API access details record
    okta_api_access_details_record = next(
        (record for record in records if record['title'] == okta_api_access_details_title), None)
    break

# Exit if Okta API credentials are not found
if okta_api_access_details_record is None:
    print(f"# Error: No Okta API Credentials record found with title: {okta_api_access_details_title}")
    exit(1)

# Extract Okta configuration details
okta_org_url = okta_api_access_details_record.get('url')
okta_org_token = okta_api_access_details_record.get('password')
old_password = params.get('oldPassword')
new_password = params.get('newPassword')

# Initialize Okta client
config = {
    "orgUrl": okta_org_url,
    "token": okta_org_token
}
okta_client = OktaClient(config)


async def get_all_users():
    """Fetch all Okta users."""
    users, _, err = await okta_client.list_users()
    if err:
        print(f"# Error: {err}")
    return users


async def get_okta_user_by_email(email):
    """Fetch an Okta user by their email."""
    print(f"# Fetching all Okta users...")
    users = await get_all_users()

    print(f"# Searching for user with email: {email}")
    found_user = next((user for user in users if user.profile.email == email), None)

    return found_user


async def rotate(user_id_to_rotate, old_password, new_password):
    """Rotate the password for a given Okta user."""

    change_pwd_request = {
        "oldPassword": old_password,
        "newPassword": new_password,
    }

    result, _, err = await okta_client.change_password(user_id_to_rotate, change_pwd_request)
    if err:
        print(f"# Error: {err}")
    else:
        print(f"# Password changed successfully for user: {user_id_to_rotate}")


async def main():
    """Main function to execute the password rotation."""
    user_email = params.get('user')

    print(f"# Fetching Okta user by email: {user_email}. This is required to get the user's Okta ID.")
    print(f"# If user id is present then there is no need to fetch the user by email and instead just use the user id.")
    user_from_okta = await get_okta_user_by_email(user_email)

    print(f"# Getting user id from Okta user")
    user_from_okta_id = user_from_okta.id if user_from_okta else None

    print(f"# Rotating password for user: {user_from_okta_id}")
    if user_from_okta_id:
        await rotate(user_from_okta_id, old_password, new_password)
    else:
        print("# Error: User {user_email} not found in Okta.")
        print("# Please ensure that the user exists in Okta and try again.")


if __name__ == '__main__':
    print("# Starting the main async function...")
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

#!/usr/local/bin/pam_rotation_venv_python3

'''
Password rotation script for Snowflake user accounts.

This script is designed to rotate the password for a given Snowflake user using the snowflake connector package.
It facilitates the automated updating of user password in your Snowflake environment.

NOTE: If spaces are present in the path to the python interpreter, the script will fail to execute.
    This is a known limitation of the shebang line in Linux and you will need to create a symlink
    to the python interpreter in a path that does not contain spaces.
    For example: sudo ln -s "/usr/local/bin/my python3.7" /usr/local/bin/pam_rotation_venv_python3
'''
import asyncio
import json
import sys
import base64

'''
Optionally display installed packages for debugging. Uncomment if needed.
import pkg_resources
print("# \n# Installed packages for debugging:")
installed_packages = pkg_resources.working_set
installed_packages_list = sorted(["%s==%s" % (i.key, i.version) for i in installed_packages])
for m in installed_packages_list:
    print(f"  {m}")
'''

# Import the snowflake connector package
try:
    import snowflake.connector
except ImportError:
    print("# Error: The 'snowflake connector' package could not be imported. Run 'pip install snowflake-connector-python' to install it.")
    exit(1)

def rotate(snowflake_account_name, snowflake_admin_user, snowflake_admin_pass, snowflake_user_name, new_password):
    """
    Connects with Snowflake using the snowflake.connector module.
    Rotate the password for a given Snowflake user.

    Args:
    - snowflake_account_name (str): The name of the Snowflake account to connect to.
    - snowflake_admin_user (str): The username of the Snowflake admin account.
    - snowflake_admin_pass (str): The password of the Snowflake admin account.
    - snowflake_user_name (str): The name of the Snowflake user whose password needs to be rotated.
    - new_password (str): The new password to be set for the Snowflake user.

    Returns:
    - None
    """

    # Connect with snowflake account using snowflake.connector module
    try:
        conn = snowflake.connector.connect(
        user=snowflake_admin_user,
        password=snowflake_admin_pass,
        account=snowflake_account_name
        )
    except Exception as E:
        print(f"Unable to connect to snowflake account. Error: {E}")
        exit(1)
    
    # Create a cursor object
    cur = conn.cursor()
    
    # Change new user's password
    try:
        change_pass_query = f"ALTER USER {snowflake_user_name} SET PASSWORD = '{new_password}'"
        cur.execute(change_pass_query)
    except Exception as E:
        print(f"Unable to update the password. Error: {E}")
        exit(1)

    # Close the cursor and connection
    cur.close()
    conn.close()

    print(f"Password successfully rotated for the given Snowflake User - {snowflake_user_name}")

def main():
    """
    Main function to rotate the password for a given Snowflake User.

    Reads and decodes input parameters from stdin, including the authentication record details
    and the new password. Then, updates the password of the specified Snowflake user.

    Args:
    - None

    Returns:
    - None
    """
    record_title = 'Snowflake Authentication Record' #This should be same as the title of the record containing admin credentials.
    admin_credential_record = None
    params = None
    
    # Read and decode input parameters from stdin
    for base64_params in sys.stdin:
        params = json.loads(base64.b64decode(base64_params).decode())
        '''
        # Optionally print available params for debugging. Uncomment if needed.
        # print(f"# \n# Available params for the script:")
        # for key, value in params.items():
        #     print(f"#     {key}={value}")
        '''

        records = json.loads(base64.b64decode(params.get('records')).decode()) # Decode and load records that are passed into the record as JSON strings in the PAM Script section as "Rotation Credential" records

        # Find the Record that contains the admin account details by its Title
        admin_credential_record = next((record for record in records if record['title'].lower() == record_title.lower()), None)
        break

    if admin_credential_record is None:
        print(f"# Error: No Record with the access token found. Title: {record_title}")
        exit(1)

    # Extract Details from the record
    snowflake_account_name = admin_credential_record.get('snowflake_account_name')
    snowflake_admin_user = admin_credential_record.get('login')
    snowflake_admin_pass = admin_credential_record.get('password')
    
    # Username for the user whose password needs to be rotated.
    snowflake_user_name = params.get('user')

    # Extract new rotated password..
    new_password = params.get('newPassword')
    
    if not all([snowflake_account_name, snowflake_admin_user, snowflake_admin_pass]):
        print("# Error: One or more required fields are missing in the authentication record.")
        exit(1)
   
    # Rotate the password for a given Snowflake user.
    rotate(snowflake_account_name, snowflake_admin_user, snowflake_admin_pass, snowflake_user_name, new_password)

if __name__ == "__main__":
    main()

#!/usr/local/bin/pam_rotation_venv_python3

# NOTE: If spaces are present in the path to the python interpreter, the script will fail to execute.
#       This is a knows limitation of the shebang line in Linux and you will need to create a symlink
#       to the python interpreter in a path that does not contain spaces.
#       For example: sudo ln -s "/usr/local/bin/my python3.7" /usr/local/bin/pam_rotation_venv_python3

# NOTE: This script is a demonstration of how to use the REST API to rotate passwords. It does not
#       actually call a real API. You will need to modify the script to call your own API.

import asyncio
import sys
import base64
import json

# Display the Python version for debugging purposes
print(f"# Python version: {sys.version}")

# Optionally display installed packages for debugging. Uncomment if needed.
# import pkg_resources
# print("# \n# Installed packages for debugging:")
# installed_packages = pkg_resources.working_set
# installed_packages_list = sorted(["%s==%s" % (i.key, i.version) for i in installed_packages])
# for m in installed_packages_list:
#     print(f"  {m}")

# Import the requests library
try:
    import requests
except ImportError:
    print("# Error: The 'requests' package is not installed. Run 'pip install requests' to install it.")
    exit(1)

# Initialize variables
api_access_token_title = 'Record with Access Token'
api_access_token_record = None
params = None

# Read and decode input parameters from stdin
for base64_params in sys.stdin:
    params = json.loads(base64.b64decode(base64_params).decode())

    # Optionally print available params for debugging. Uncomment if needed.
    # print(f"# \n# Available params for the script:")
    # for key, value in params.items():
    #     print(f"#     {key}={value}")

    # Decode and load records that are passed into the record as JSON strings 
    # in the PAM Script section as "Rotation Credential" records
    records = json.loads(base64.b64decode(params.get('records')).decode())

    # Find the Record that contains the access token by its Title
    api_access_token_record = next(
        (record for record in records if record['title'] == api_access_token_title), None)
    break

if api_access_token_record is None:
    print(f"# Error: No Record with the access token found. Title: {api_access_token_title}")
    exit(1)

# Extract Details from the record details
rest_api_service_url = api_access_token_record.get('url')
service_access_token = api_access_token_record.get('password')

# Extract details from the record
rest_api_service_url = api_access_token_record.get('url')
service_access_token = api_access_token_record.get('password')
# old_password = params.get('oldPassword')  # in case if you want to use previous password
new_password = params.get('newPassword')
user_email = params.get('user')             # user of the account for which password should be rotated

def rotate(user_email, new_password):
    """
    Rotate the password for a given user.

    Args:
    - user_email (str): The email of the user for whom the password should be rotated.
    - new_password (str): The new password.

    Returns:
    - dict: The response from the API call.
    """

    # Headers for the request
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {service_access_token}"
    }

    # Prepare the request payload
    request_payload = {
        "userEmail": user_email,
        "newPassword": new_password,
    }

    # Make a POST request to the API
    response = requests.post(url=rest_api_service_url, headers=headers, data=json.dumps(request_payload))

    # Check if the request was successful
    if response.status_code == 201:
        return response.json()
    else:
        response.raise_for_status()

if __name__ == '__main__':
    """Main function to execute the password rotation."""

    response = rotate(user_email, new_password)

    # Print the response for debugging purposes
    print(f"# \n# Response from the API:")
    print(f"# {response}")