random.py

# Individual code blocks that have security issues and/or security concerns #
#############################################################################
# Block 1 #

import subprocess
import requests
import logging
import json

#############################################################################
# Block 2

def validate_file(filename):
    command = 'validator -i "{filename}"'.format(filename=filename)
    subprocess.call(command, shell=True)

#############################################################################
# Block 3
bearer = "asdf-123456"


def admin_access_check(user):
    assert user.is_admin, "user does not have access"
    url = f"http://ops.example.com/users/{user}/details"
    headers = {
        "Authorization": bearer,
        "Content-type": "application/json",
        "Accept": "application/json"
    }
    requests.get(url, headers=headers)
    return

#############################################################################
# Block 4

bearer = "asdf-123456"


def cluster_creation(url, cluster_name):
    payload = {
        "dockerRootDir": "/var/lib/docker",
        "enableNetworkPolicy": "false",
        "type": "cluster",
        "name": cluster_name,
        "EngineConfig": {
            "addonJobTimeout": 30,
            "ignoreDockerVersion": "true",
            "sshAgentAuth": "false",
            "type": "EngineConfig",
            "authentication": {"type": "authConfig", "strategy": "x509"},
            "network": {"type": "networkConfig", "plugin": "canal"},
            "ingress": {"type": "ingressConfig", "provider": "none"},
            "monitoring": {"type": "monitoringConfig", "provider": "metrics"},
            "services": {
                "type": "ConfigServices",
                "kubeApi": {"podSecurityPolicy": "true"},
                "etcd": {
                    "snapshot": "false",
                    "type": "etcdService"
                },
            },
        },
    }
    try:
        headers = {
            "Authorization": bearer,
            "Content-type": "application/json",
            "Accept": "application/json",
        }
        resp = requests.post(url, data=json.dumps(payload), headers=headers)
        resp.raise_for_status()
        if resp.status_code == 201:
            logging.info("Cluster created successfully!")
        elif resp.status_code == 409:
            logging.info("Cluster already exists!")
        elif resp.status_code == 422:
            logging.error("Cannot process this request")
            exit(0)
    except requests.exceptions.ConnectionError:
        logging.error("Connection error to URL: " + url)
        exit(0)
    except requests.exceptions.Timeout:
        logging.error("Timeout error")
        exit(0)
    except requests.exceptions.TooManyRedirects:
        logging.error("Too many redirects")
        exit(0)
    except requests.exceptions.RequestException as e:
        logging.error(e)
        exit(1)

#############################################################################
# Block 5

class AdminSuperClass():
    def __init__(self):
        self.__private = 1

    def get_private(self):
        return self.__private

    def has_private(self):
        return hasattr(self, '__private')