13595221517

Committed 28 Feb 2025 07:34PM UTC coverage: 83.711%. First build

Build # 13595221517

Build Type

Pull #348

github

Committed by

web-flow

Commit Message

Merge 3c6efe1fa into ee047d77d

Pull Request Pull Request #348: Fix for #346 & #347

Run Details

17 of 19 new or added lines in 3 files covered. (89.47%)

2292 of 2738 relevant lines covered (83.71%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.11

/tabcmd/commands/auth/session.py

import getpass
import json
import os

import requests
import tableauserverclient as TSC
import urllib3
from urllib3.exceptions import InsecureRequestWarning
from urllib.parse import urlparse, urlunparse

from tabcmd.version import version
from tabcmd.commands.constants import Errors
from tabcmd.execution.localize import _
from tabcmd.execution.logger_config import log

from typing import Dict, Any


class Session:
    """
    Session class handles all authentication related work
    """

    TOKEN_CRED_TYPE = "token"
    PASSWORD_CRED_TYPE = "password"

    TOKEN_CRED_TYPE = "token"
    PASSWORD_CRED_TYPE = "password"

    def __init__(self):
        self.username = None
        # we don't store the password
        self.user_id = None
        self.auth_token = None
        self.token_name = None
        self.token_value = None
        self.password_file = None
        self.token_file = None
        self.site_name = None  # The site name, e.g 'alpodev'
        self.site_id = None  # The site id, e.g 'abcd-1234-1234-1244-1234'
        self.server_url = None
        self.last_command = None  # for when we have to renew the session then re-try
        self.last_login_using = None

        self.no_prompt = False
        self.certificate = None
        self.no_certcheck = False
        self.no_proxy = False
        self.proxy = None
        self.timeout = None

        self.logging_level = "info"
        self.logger = log(__name__, self.logging_level)  # instantiate here mostly for tests
        self._read_from_json()
        self.tableau_server = None  # this one is an object that doesn't get persisted in the file

    # called before we connect to the server
    # generally, we don't want to overwrite stored data with nulls
    def _update_session_data(self, args):
        # user id and site id are never passed in as args
        # last_login_using and tableau_server are internal data
        # self.command = args.???
        self.username = args.username or self.username or ""
        self.username = self.username.lower()
        server_base_url = self._get_server_base_url(args.server) if args.server else None
        self.server_url = server_base_url or self.server_url or "http://localhost"
        self.server_url = self.server_url.lower()
        if args.server is not None:
            self.site_name = None
        self.site_name = args.site_name or self.site_name or ""
        self.site_name = self.site_name.lower()
        if self.site_name == "default":
            self.site_name = ""
        self.logging_level = args.logging_level or self.logging_level
        self.password_file = args.password_file or self.password_file
        self.token_file = args.token_file or self.token_file
        self.token_name = args.token_name or self.token_name
        self.token_value = args.token_value or self.token_value

        self.no_prompt = args.no_prompt  # have to set this on every call?
        self.certificate = args.certificate or self.certificate
        self.no_certcheck = args.no_certcheck  # have to set this on every call?
        self.no_proxy = args.no_proxy  # have to set this on every call?
        self.proxy = args.proxy or self.proxy
        self.timeout = self.timeout_as_integer(self.logger, args.timeout, self.timeout)

    @staticmethod
    def timeout_as_integer(logger, option_1, option_2):
        result = None
        if option_1:
            try:
                result = int(option_1)
            except Exception as anyE:
                result = 0
        if option_2 and (not result or result <= 0):
            try:
                result = int(option_2)
            except Exception as anyE:
                result = 0
        if not option_1 and not option_2:
            logger.debug(_("setsetting.status").format("timeout", "None"))
        elif not result or result <= 0:
            logger.warning(_("sessionoptions.errors.bad_timeout").format("--timeout", result))
        return result or 0

    @staticmethod
    def _read_password_from_file(filename):
        credential = None
        with open(str(filename), "r") as file_contents:
            reader = file_contents.readlines()
            for row in reader:
                credential = row
            return credential

    def _allow_prompt(self):
        try:
            return not self.no_prompt
        except Exception:
            return True

    def _create_new_credential(self, password, credential_type):
        if password is None:
            if self.password_file:
                password = Session._read_password_from_file(self.password_file)
            elif self._allow_prompt():
                password = getpass.getpass(_("session.password"))
            else:
                Errors.exit_with_error(self.logger, _("session.errors.script_no_password"))

        if credential_type == Session.PASSWORD_CRED_TYPE and self.username and password:
            credentials = TSC.TableauAuth(self.username, password, site_id=self.site_name)
            self.last_login_using = "username"
            return credentials
        elif credential_type == Session.TOKEN_CRED_TYPE and self.token_name:
            credentials = self._create_new_token_credential()
            return credentials
        else:
            Errors.exit_with_error(self.logger, _("session.errors.missing_arguments").format(""))

    def _create_new_token_credential(self):
        if self.token_value:
            token = self.token_value
        elif self.token_file:
            token = Session._read_password_from_file(self.token_file)
        elif self._allow_prompt():
            token = getpass.getpass("Token:")
        else:
            Errors.exit_with_error(self.logger, _("session.errors.missing_arguments").format("token"))

        if self.token_name and token:
            credentials = TSC.PersonalAccessTokenAuth(self.token_name, token, site_id=self.site_name)
            self.last_login_using = "token"
            return credentials
        else:
            Errors.exit_with_error(self.logger, _("session.errors.missing_arguments").format("token name"))

    def _open_connection_with_opts(self) -> TSC.Server:
        self.logger.debug("Setting up request options")
        http_options: Dict[str, Any] = {"headers": {"User-Agent": "Tabcmd/{}".format(version)}}

        if self.no_certcheck:
            http_options["verify"] = False
            urllib3.disable_warnings(category=InsecureRequestWarning)

        """
           Do we want to do the same format check as old tabcmd?
           For now I think we can trust requests to handle a bad proxy
           Pattern pattern = Pattern.compile("([^:]*):([0-9]*)");           
           if not matches:
               throw new ReportableException(m_i18n.getString("sessionoptions.errors.bad_proxy_format", proxyArg));
        """
        if self.proxy:
            self.logger.debug("Setting http proxy: {}".format(self.proxy))
            proxies = {"http": self.proxy}
            http_options["proxies"] = proxies
        if self.no_proxy:
            # override any proxy that was set
            http_options["proxies"] = None

        if self.timeout:
            http_options["timeout"] = self.timeout

        if self.certificate:
            http_options["cert"] = self.certificate

        try:
            self.logger.debug(http_options)
            # this is the only place we open a connection to the server
            # so the request options are all set for the session now
            tableau_server = TSC.Server(self.server_url, http_options=http_options)

        except Exception as e:
            self.logger.debug(
                "Connection args: server {}, site {}, proxy {}/no-proxy {}, cert {}".format(
                    self.server_url, self.site_name, self.proxy, self.no_proxy, self.certificate
                )
            )
            Errors.exit_with_error(self.logger, "Failed to connect to server", e)

        self.logger.debug("Finished setting up connection")
        return tableau_server

    def _verify_server_connection_unauthed(self):
        try:
            self.tableau_server.use_server_version()
        except requests.exceptions.ReadTimeout as timeout_error:
            Errors.exit_with_error(
                self.logger,
                message="Timed out after {} seconds attempting to connect to server".format(self.timeout),
                exception=timeout_error,
            )
        except requests.exceptions.RequestException as requests_error:
            Errors.exit_with_error(
                self.logger, message="Error attempting to connect to the server", exception=requests_error
            )
        except Exception as e:
            Errors.exit_with_error(self.logger, exception=e)

    def _create_new_connection(self) -> TSC.Server:
        self._print_server_info()
        self.logger.info(_("session.connecting"))
        try:
            self.tableau_server = self._open_connection_with_opts()
        except Exception as e:
            Errors.exit_with_error(self.logger, "Failed to connect to server", e)
        return self.tableau_server

    def _read_existing_state(self):
        if self._json_exists():
            self._read_from_json()

    def _print_server_info(self):
        self.logger.info("=====   Server: {}".format(self.server_url))
        if self.proxy:
            self.logger.info("=====   Proxy: {}".format(self.proxy))
        if self.username:
            self.logger.info("=====   Username: {}".format(self.username))
        if self.certificate:
            self.logger.info("=====   Certificate: {}".format(self.certificate))
        else:
            self.logger.info("=====   Token Name: {}".format(self.token_name))
        site_display_name = self.site_name or "Default Site"
        self.logger.info(_("dataconnections.classes.tableau_server_site") + ": {}".format(site_display_name))

    # side-effect: sets self.username
    def _validate_existing_signin(self):
        # when do these two messages show up? self.logger.info(_("session.auto_site_login"))
        try:
            if self.tableau_server and self.tableau_server.is_signed_in():
                server_user = self.tableau_server.users.get_by_id(self.user_id).name
                if not self.username:
                    self.logger.info("Fetched user details from server")
                    self.username = server_user

                return self.tableau_server
        except TSC.ServerResponseError as e:
            self.logger.info(_("publish.errors.unexpected_server_response"), e)
        except Exception as e:
            self.logger.info(_("errors.internal_error.request.message"), e)
        return None

    # server connection created, not yet logged in
    def _sign_in(self, tableau_auth) -> TSC.Server:
        self.logger.debug(_("session.login") + self.server_url)
        self.logger.debug(_("listsites.output").format("", self.username or self.token_name, self.site_name))
        try:
            self.tableau_server.auth.sign_in(tableau_auth)  # it's the same call for token or user-pass
        except Exception as e:
            Errors.exit_with_error(self.logger, exception=e)
        try:
            self.site_id = self.tableau_server.site_id
            self.user_id = self.tableau_server.user_id
            self.auth_token = self.tableau_server._auth_token
            success = self._validate_existing_signin()
        except Exception as e:
            Errors.exit_with_error(self.logger, exception=e)
        if success:
            self.logger.info(_("common.output.succeeded"))
        else:
            Errors.exit_with_error(self.logger, message="Sign in failed")

        return self.tableau_server

    def _get_saved_credentials(self):
        if self.last_login_using == "username":
            credentials = self._create_new_credential(None, Session.PASSWORD_CRED_TYPE)
        elif self.last_login_using == "token":
            credentials = self._create_new_token_credential()
        else:
            return None

        return credentials

    # external entry point:
    def create_session(self, args, logger):
        signed_in_object = None
        # pull out cached info from json, then overwrite with new args if available
        self._read_existing_state()
        self._update_session_data(args)
        self.logging_level = args.logging_level or self.logging_level
        self.logger = logger or log(__class__.__name__, self.logging_level)

        credentials = None
        if args.password or args.password_file:
            self._end_session()
            # we don't save the password anywhere, so we pass it along directly
            credentials = self._create_new_credential(args.password, Session.PASSWORD_CRED_TYPE)
        elif args.token_value or args.token_file:
            self._end_session()
            credentials = self._create_new_token_credential()
        else:  # no login arguments given - look for saved info
            # maybe we're already signed in!
            if self.tableau_server:
                self.logger.info(_("session.continuing_session"))
                signed_in_object = self._validate_existing_signin()

            if not signed_in_object:
                credentials = self._get_saved_credentials()

        if credentials and not signed_in_object:
            self.logger.debug("Signin details found:")
            self.tableau_server = self._create_new_connection()
            self._verify_server_connection_unauthed()
            signed_in_object = self._sign_in(credentials)

        if not signed_in_object:
            message = "Run 'tabcmd login -h' for details on required arguments"
            Errors.exit_with_error(self.logger, _("session.errors.missing_arguments").format(message))
        if args.no_cookie:
            self._remove_json()
        else:
            self._save_session_to_json()
        return signed_in_object

    def end_session_and_clear_data(self):
        self._end_session()
        self.logger.info(_("session.logout"))
        self._clear_data()

    def _end_session(self):
        if self.tableau_server:
            self.tableau_server.auth.sign_out()
            self.tableau_server = None

    def _clear_data(self):
        self._remove_json()
        self.username = None
        self.user_id = None
        self.auth_token = None
        self.token_name = None
        self.token_value = None
        self.site_name = None
        self.site_id = None
        self.server = None
        self.last_login_using = None
        self.password_file = None
        self.token_file = None

        self.last_command = None
        self.tableau_server = None

        self.certificate = None
        self.no_certcheck = None
        self.no_proxy = None
        self.proxy = None
        self.timeout = None

    # json file functions ----------------------------------------------------
    # These should be moved into a separate class
    def _get_file_path(self):
        home_path = os.path.expanduser("~")
        file_path = os.path.join(home_path, "tableau_auth.json")
        return file_path

    def _read_from_json(self):
        if not self._json_exists():
            return
        file_path = self._get_file_path()
        content = None
        try:
            with open(str(file_path), "r") as file_contents:
                data = json.load(file_contents)
                if data is None or data == {}:
                    return
                content = data["tableau_auth"]
                if content is None:
                    return
                self._save_data_from_json(content)
        except json.JSONDecodeError as e:
            self._wipe_bad_json(e, "Error reading data from session file")
        except IOError as e:
            self._wipe_bad_json(e, "Error reading session file")
        except AttributeError as e:
            self._wipe_bad_json(e, "Error parsing session details from file")
        except Exception as e:
            self._wipe_bad_json(e, "Unexpected error reading session details from file")

    def _save_data_from_json(self, content):
        try:
            auth = content[0]
            if auth is None:
                self._wipe_bad_json(ValueError(), "Empty session file")
            self.auth_token = auth["auth_token"]
            self.server_url = auth["server"]
            self.site_name = auth["site_name"]
            self.site_id = auth["site_id"]
            self.username = auth["username"]
            self.user_id = auth["user_id"]
            self.token_name = auth["personal_access_token_name"]
            self.token_value = auth["personal_access_token"]
            self.last_login_using = auth["last_login_using"]
            self.password_file = auth["password_file"]
            self.token_file = auth["token_file"]
            self.no_prompt = auth["no_prompt"]
            self.no_certcheck = auth["no_certcheck"]
            self.certificate = auth["certificate"]
            self.no_proxy = auth["no_proxy"]
            self.proxy = auth["proxy"]
            self.timeout = auth["timeout"]
        except AttributeError as e:
            self._wipe_bad_json(e, "Unrecognized attribute in session file")
        except Exception as e:
            self._wipe_bad_json(e, "Failed to load session file")

    def _wipe_bad_json(self, e, message):
        self.logger.debug(message + ": " + e.__str__())
        self.logger.info(_("session.new_session"))
        self._remove_json()

    def _json_exists(self):
        # todo: make this location configurable
        home_path = os.path.expanduser("~")
        file_path = os.path.join(home_path, "tableau_auth.json")
        return os.path.exists(file_path)

    def _save_session_to_json(self):
        try:
            data = self._serialize_for_save()
            self._save_file(data)
        except Exception as e:
            self._wipe_bad_json(e, "Failed to save session file")

    def _save_file(self, data):
        file_path = self._get_file_path()
        with open(str(file_path), "w") as f:
            json.dump(data, f)

    def _serialize_for_save(self):
        data = {"tableau_auth": []}
        data["tableau_auth"].append(
            {
                "auth_token": self.auth_token,
                "server": self.server_url,
                "username": self.username,
                "user_id": self.user_id,
                "site_name": self.site_name,
                "site_id": self.site_id,
                "personal_access_token_name": self.token_name,
                "personal_access_token": self.token_value,
                "last_login_using": self.last_login_using,
                "password_file": self.password_file,
                "token_file": self.token_file,
                "no_prompt": self.no_prompt,
                "no_certcheck": self.no_certcheck,
                "certificate": self.certificate,
                "no_proxy": self.no_proxy,
                "proxy": self.proxy,
                "timeout": self.timeout,
            }
        )
        return data

    def _remove_json(self):
        file_path = ""
        try:
            if not self._json_exists():
                return
            file_path = self._get_file_path()
            self._save_file({})
            if os.path.exists(file_path):
                os.remove(file_path)
        except Exception as e:
            message = "Error clearing session data from {}: check and remove manually".format(file_path)
            self.logger.error(message)
            self.logger.error(e)

    def _get_server_base_url(self, url: str):
        try:
            parsed = urlparse(url)
            scheme = parsed.scheme or "http"

            # If netloc is empty, treat path as netloc and discard any extra path
            netloc = parsed.netloc or parsed.path.split("/")[0]  # Keep only the domain
            return urlunparse((scheme, netloc, "", "", "", ""))
        except Exception as e:
            Errors.exit_with_error(self.logger, exception=e)

1	import getpass	1✔
2	import json	1✔
3	import os	1✔
4
5	import requests	1✔
6	import tableauserverclient as TSC	1✔
7	import urllib3	1✔
8	from urllib3.exceptions import InsecureRequestWarning	1✔
9	from urllib.parse import urlparse, urlunparse	1✔
10
11	from tabcmd.version import version	1✔
12	from tabcmd.commands.constants import Errors	1✔
13	from tabcmd.execution.localize import _	1✔
14	from tabcmd.execution.logger_config import log	1✔
15
16	from typing import Dict, Any	1✔
17
18
19	class Session:	1✔
20	"""
21	Session class handles all authentication related work
22	"""
23
24	TOKEN_CRED_TYPE = "token"	1✔
25	PASSWORD_CRED_TYPE = "password"	1✔
26
27	TOKEN_CRED_TYPE = "token"	1✔
28	PASSWORD_CRED_TYPE = "password"	1✔
29
30	def __init__(self):	1✔
31	self.username = None	1✔
32	# we don't store the password
33	self.user_id = None	1✔
34	self.auth_token = None	1✔
35	self.token_name = None	1✔
36	self.token_value = None	1✔
37	self.password_file = None	1✔
38	self.token_file = None	1✔
39	self.site_name = None # The site name, e.g 'alpodev'	1✔
40	self.site_id = None # The site id, e.g 'abcd-1234-1234-1244-1234'	1✔
41	self.server_url = None	1✔
42	self.last_command = None # for when we have to renew the session then re-try	1✔
43	self.last_login_using = None	1✔
44
45	self.no_prompt = False	1✔
46	self.certificate = None	1✔
47	self.no_certcheck = False	1✔
48	self.no_proxy = False	1✔
49	self.proxy = None	1✔
50	self.timeout = None	1✔
51
52	self.logging_level = "info"	1✔
53	self.logger = log(__name__, self.logging_level) # instantiate here mostly for tests	1✔
54	self._read_from_json()	1✔
55	self.tableau_server = None # this one is an object that doesn't get persisted in the file	1✔
56
57	# called before we connect to the server
58	# generally, we don't want to overwrite stored data with nulls
59	def _update_session_data(self, args):	1✔
60	# user id and site id are never passed in as args
61	# last_login_using and tableau_server are internal data
62	# self.command = args.???
63	self.username = args.username or self.username or ""	1✔
64	self.username = self.username.lower()	1✔
65	server_base_url = self._get_server_base_url(args.server) if args.server else None	1✔
66	self.server_url = server_base_url or self.server_url or "http://localhost"	1✔
67	self.server_url = self.server_url.lower()	1✔
68	if args.server is not None:	1✔
69	self.site_name = None	1✔
70	self.site_name = args.site_name or self.site_name or ""	1✔
71	self.site_name = self.site_name.lower()	1✔
72	if self.site_name == "default":	1✔
73	self.site_name = ""	×
74	self.logging_level = args.logging_level or self.logging_level	1✔
75	self.password_file = args.password_file or self.password_file	1✔
76	self.token_file = args.token_file or self.token_file	1✔
77	self.token_name = args.token_name or self.token_name	1✔
78	self.token_value = args.token_value or self.token_value	1✔
79
80	self.no_prompt = args.no_prompt # have to set this on every call?	1✔
81	self.certificate = args.certificate or self.certificate	1✔
82	self.no_certcheck = args.no_certcheck # have to set this on every call?	1✔
83	self.no_proxy = args.no_proxy # have to set this on every call?	1✔
84	self.proxy = args.proxy or self.proxy	1✔
85	self.timeout = self.timeout_as_integer(self.logger, args.timeout, self.timeout)	1✔
86
87	@staticmethod	1✔
88	def timeout_as_integer(logger, option_1, option_2):	1✔
89	result = None	1✔
90	if option_1:	1✔
91	try:	1✔
92	result = int(option_1)	1✔
93	except Exception as anyE:	1✔
94	result = 0	1✔
95	if option_2 and (not result or result <= 0):	1✔
96	try:	1✔
97	result = int(option_2)	1✔
98	except Exception as anyE:	×
99	result = 0	×
100	if not option_1 and not option_2:	1✔
101	logger.debug(_("setsetting.status").format("timeout", "None"))	1✔
102	elif not result or result <= 0:	1✔
103	logger.warning(_("sessionoptions.errors.bad_timeout").format("--timeout", result))	1✔
104	return result or 0	1✔
105
106	@staticmethod	1✔
107	def _read_password_from_file(filename):	1✔
108	credential = None	1✔
109	with open(str(filename), "r") as file_contents:	1✔
110	reader = file_contents.readlines()	1✔
111	for row in reader:	1✔
112	credential = row	1✔
113	return credential	1✔
114
115	def _allow_prompt(self):	1✔
116	try:	1✔
117	return not self.no_prompt	1✔
118	except Exception:	×
119	return True	×
120
121	def _create_new_credential(self, password, credential_type):	1✔
122	if password is None:	1✔
123	if self.password_file:	1✔
124	password = Session._read_password_from_file(self.password_file)	1✔
125	elif self._allow_prompt():	1✔
126	password = getpass.getpass(_("session.password"))	1✔
127	else:
128	Errors.exit_with_error(self.logger, _("session.errors.script_no_password"))	×
129
130	if credential_type == Session.PASSWORD_CRED_TYPE and self.username and password:	1✔
131	credentials = TSC.TableauAuth(self.username, password, site_id=self.site_name)	1✔
132	self.last_login_using = "username"	1✔
133	return credentials	1✔
134	elif credential_type == Session.TOKEN_CRED_TYPE and self.token_name:	1✔
135	credentials = self._create_new_token_credential()	×
136	return credentials	×
137	else:
138	Errors.exit_with_error(self.logger, _("session.errors.missing_arguments").format(""))	1✔
139
140	def _create_new_token_credential(self):	1✔
141	if self.token_value:	1✔
142	token = self.token_value	1✔
143	elif self.token_file:	1✔
144	token = Session._read_password_from_file(self.token_file)	1✔
145	elif self._allow_prompt():	1✔
146	token = getpass.getpass("Token:")	1✔
147	else:
148	Errors.exit_with_error(self.logger, _("session.errors.missing_arguments").format("token"))	×
149
150	if self.token_name and token:	1✔
151	credentials = TSC.PersonalAccessTokenAuth(self.token_name, token, site_id=self.site_name)	1✔
152	self.last_login_using = "token"	1✔
153	return credentials	1✔
154	else:
155	Errors.exit_with_error(self.logger, _("session.errors.missing_arguments").format("token name"))	1✔
156
157	def _open_connection_with_opts(self) -> TSC.Server:	1✔
158	self.logger.debug("Setting up request options")	1✔
159	http_options: Dict[str, Any] = {"headers": {"User-Agent": "Tabcmd/{}".format(version)}}	1✔
160
161	if self.no_certcheck:	1✔
162	http_options["verify"] = False	1✔
163	urllib3.disable_warnings(category=InsecureRequestWarning)	1✔
164
165	"""	1✔
166	Do we want to do the same format check as old tabcmd?
167	For now I think we can trust requests to handle a bad proxy
168	Pattern pattern = Pattern.compile("([^:]):([0-9])");
169	if not matches:
170	throw new ReportableException(m_i18n.getString("sessionoptions.errors.bad_proxy_format", proxyArg));
171	"""
172	if self.proxy:	1✔
173	self.logger.debug("Setting http proxy: {}".format(self.proxy))	1✔
174	proxies = {"http": self.proxy}	1✔
175	http_options["proxies"] = proxies	1✔
176	if self.no_proxy:	1✔
177	# override any proxy that was set
178	http_options["proxies"] = None	1✔
179
180	if self.timeout:	1✔
181	http_options["timeout"] = self.timeout	1✔
182
183	if self.certificate:	1✔
184	http_options["cert"] = self.certificate	1✔
185
186	try:	1✔
187	self.logger.debug(http_options)	1✔
188	# this is the only place we open a connection to the server
189	# so the request options are all set for the session now
190	tableau_server = TSC.Server(self.server_url, http_options=http_options)	1✔
191
192	except Exception as e:	×
193	self.logger.debug(	×
194	"Connection args: server {}, site {}, proxy {}/no-proxy {}, cert {}".format(
195	self.server_url, self.site_name, self.proxy, self.no_proxy, self.certificate
196	)
197	)
198	Errors.exit_with_error(self.logger, "Failed to connect to server", e)	×
199
200	self.logger.debug("Finished setting up connection")	1✔
201	return tableau_server	1✔
202
203	def _verify_server_connection_unauthed(self):	1✔
204	try:	1✔
205	self.tableau_server.use_server_version()	1✔
206	except requests.exceptions.ReadTimeout as timeout_error:	×
207	Errors.exit_with_error(	×
208	self.logger,
209	message="Timed out after {} seconds attempting to connect to server".format(self.timeout),
210	exception=timeout_error,
211	)
212	except requests.exceptions.RequestException as requests_error:	×
213	Errors.exit_with_error(	×
214	self.logger, message="Error attempting to connect to the server", exception=requests_error
215	)
216	except Exception as e:	×
217	Errors.exit_with_error(self.logger, exception=e)	×
218
219	def _create_new_connection(self) -> TSC.Server:	1✔
220	self._print_server_info()	1✔
221	self.logger.info(_("session.connecting"))	1✔
222	try:	1✔
223	self.tableau_server = self._open_connection_with_opts()	1✔
224	except Exception as e:	×
225	Errors.exit_with_error(self.logger, "Failed to connect to server", e)	×
226	return self.tableau_server	1✔
227
228	def _read_existing_state(self):	1✔
229	if self._json_exists():	1✔
230	self._read_from_json()	1✔
231
232	def _print_server_info(self):	1✔
233	self.logger.info("===== Server: {}".format(self.server_url))	1✔
234	if self.proxy:	1✔
235	self.logger.info("===== Proxy: {}".format(self.proxy))	1✔
236	if self.username:	1✔
237	self.logger.info("===== Username: {}".format(self.username))	1✔
238	if self.certificate:	1✔
239	self.logger.info("===== Certificate: {}".format(self.certificate))	1✔
240	else:
241	self.logger.info("===== Token Name: {}".format(self.token_name))	1✔
242	site_display_name = self.site_name or "Default Site"	1✔
243	self.logger.info(_("dataconnections.classes.tableau_server_site") + ": {}".format(site_display_name))	1✔
244
245	# side-effect: sets self.username
246	def _validate_existing_signin(self):	1✔
247	# when do these two messages show up? self.logger.info(_("session.auto_site_login"))
248	try:	1✔
249	if self.tableau_server and self.tableau_server.is_signed_in():	1✔
250	server_user = self.tableau_server.users.get_by_id(self.user_id).name	1✔
251	if not self.username:	1✔
252	self.logger.info("Fetched user details from server")	1✔
253	self.username = server_user	1✔
254
255	return self.tableau_server	1✔
256	except TSC.ServerResponseError as e:	×
257	self.logger.info(_("publish.errors.unexpected_server_response"), e)	×
258	except Exception as e:	×
259	self.logger.info(_("errors.internal_error.request.message"), e)	×
260	return None	×
261
262	# server connection created, not yet logged in
263	def _sign_in(self, tableau_auth) -> TSC.Server:	1✔
264	self.logger.debug(_("session.login") + self.server_url)	1✔
265	self.logger.debug(_("listsites.output").format("", self.username or self.token_name, self.site_name))	1✔
266	try:	1✔
267	self.tableau_server.auth.sign_in(tableau_auth) # it's the same call for token or user-pass	1✔
268	except Exception as e:	1✔
269	Errors.exit_with_error(self.logger, exception=e)	1✔
270	try:	1✔
271	self.site_id = self.tableau_server.site_id	1✔
272	self.user_id = self.tableau_server.user_id	1✔
273	self.auth_token = self.tableau_server._auth_token	1✔
274	success = self._validate_existing_signin()	1✔
275	except Exception as e:	×
276	Errors.exit_with_error(self.logger, exception=e)	×
277	if success:	1✔
278	self.logger.info(_("common.output.succeeded"))	1✔
279	else:
280	Errors.exit_with_error(self.logger, message="Sign in failed")	×
281
282	return self.tableau_server	1✔
283
284	def _get_saved_credentials(self):	1✔
285	if self.last_login_using == "username":	1✔
286	credentials = self._create_new_credential(None, Session.PASSWORD_CRED_TYPE)	×
287	elif self.last_login_using == "token":	1✔
288	credentials = self._create_new_token_credential()	×
289	else:
290	return None	1✔
291
292	return credentials	×
293
294	# external entry point:
295	def create_session(self, args, logger):	1✔
296	signed_in_object = None	1✔
297	# pull out cached info from json, then overwrite with new args if available
298	self._read_existing_state()	1✔
299	self._update_session_data(args)	1✔
300	self.logging_level = args.logging_level or self.logging_level	1✔
301	self.logger = logger or log(__class__.__name__, self.logging_level)	1✔
302
303	credentials = None	1✔
304	if args.password or args.password_file:	1✔
305	self._end_session()	1✔
306	# we don't save the password anywhere, so we pass it along directly
307	credentials = self._create_new_credential(args.password, Session.PASSWORD_CRED_TYPE)	1✔
308	elif args.token_value or args.token_file:	1✔
309	self._end_session()	1✔
310	credentials = self._create_new_token_credential()	1✔
311	else: # no login arguments given - look for saved info
312	# maybe we're already signed in!
313	if self.tableau_server:	1✔
314	self.logger.info(_("session.continuing_session"))	×
315	signed_in_object = self._validate_existing_signin()	×
316
317	if not signed_in_object:	1✔
318	credentials = self._get_saved_credentials()	1✔
319
320	if credentials and not signed_in_object:	1✔
321	self.logger.debug("Signin details found:")	1✔
322	self.tableau_server = self._create_new_connection()	1✔
323	self._verify_server_connection_unauthed()	1✔
324	signed_in_object = self._sign_in(credentials)	1✔
325
326	if not signed_in_object:	1✔
327	message = "Run 'tabcmd login -h' for details on required arguments"	1✔
328	Errors.exit_with_error(self.logger, _("session.errors.missing_arguments").format(message))	1✔
329	if args.no_cookie:	1✔
330	self._remove_json()	×
331	else:
332	self._save_session_to_json()	1✔
333	return signed_in_object	1✔
334
335	def end_session_and_clear_data(self):	1✔
336	self._end_session()	1✔
337	self.logger.info(_("session.logout"))	1✔
338	self._clear_data()	1✔
339
340	def _end_session(self):	1✔
341	if self.tableau_server:	1✔
342	self.tableau_server.auth.sign_out()	1✔
343	self.tableau_server = None	1✔
344
345	def _clear_data(self):	1✔
346	self._remove_json()	1✔
347	self.username = None	1✔
348	self.user_id = None	1✔
349	self.auth_token = None	1✔
350	self.token_name = None	1✔
351	self.token_value = None	1✔
352	self.site_name = None	1✔
353	self.site_id = None	1✔
354	self.server = None	1✔
355	self.last_login_using = None	1✔
356	self.password_file = None	1✔
357	self.token_file = None	1✔
358
359	self.last_command = None	1✔
360	self.tableau_server = None	1✔
361
362	self.certificate = None	1✔
363	self.no_certcheck = None	1✔
364	self.no_proxy = None	1✔
365	self.proxy = None	1✔
366	self.timeout = None	1✔
367
368	# json file functions ----------------------------------------------------
369	# These should be moved into a separate class
370	def _get_file_path(self):	1✔
371	home_path = os.path.expanduser("~")	1✔
372	file_path = os.path.join(home_path, "tableau_auth.json")	1✔
373	return file_path	1✔
374
375	def _read_from_json(self):	1✔
376	if not self._json_exists():	1✔
377	return	1✔
378	file_path = self._get_file_path()	1✔
379	content = None	1✔
380	try:	1✔
381	with open(str(file_path), "r") as file_contents:	1✔
382	data = json.load(file_contents)	1✔
383	if data is None or data == {}:	1✔
384	return	1✔
385	content = data["tableau_auth"]	1✔
386	if content is None:	1✔
387	return	×
388	self._save_data_from_json(content)	1✔
389	except json.JSONDecodeError as e:	1✔
390	self._wipe_bad_json(e, "Error reading data from session file")	×
391	except IOError as e:	1✔
392	self._wipe_bad_json(e, "Error reading session file")	×
393	except AttributeError as e:	1✔
394	self._wipe_bad_json(e, "Error parsing session details from file")	×
395	except Exception as e:	1✔
396	self._wipe_bad_json(e, "Unexpected error reading session details from file")	1✔
397
398	def _save_data_from_json(self, content):	1✔
399	try:	1✔
400	auth = content[0]	1✔
401	if auth is None:	1✔
402	self._wipe_bad_json(ValueError(), "Empty session file")	×
403	self.auth_token = auth["auth_token"]	1✔
404	self.server_url = auth["server"]	1✔
405	self.site_name = auth["site_name"]	1✔
406	self.site_id = auth["site_id"]	1✔
407	self.username = auth["username"]	1✔
408	self.user_id = auth["user_id"]	1✔
409	self.token_name = auth["personal_access_token_name"]	1✔
410	self.token_value = auth["personal_access_token"]	1✔
411	self.last_login_using = auth["last_login_using"]	1✔
412	self.password_file = auth["password_file"]	1✔
413	self.token_file = auth["token_file"]	1✔
414	self.no_prompt = auth["no_prompt"]	1✔
415	self.no_certcheck = auth["no_certcheck"]	1✔
416	self.certificate = auth["certificate"]	1✔
417	self.no_proxy = auth["no_proxy"]	1✔
418	self.proxy = auth["proxy"]	1✔
419	self.timeout = auth["timeout"]	1✔
420	except AttributeError as e:	1✔
421	self._wipe_bad_json(e, "Unrecognized attribute in session file")	×
422	except Exception as e:	1✔
423	self._wipe_bad_json(e, "Failed to load session file")	1✔
424
425	def _wipe_bad_json(self, e, message):	1✔
426	self.logger.debug(message + ": " + e.__str__())	1✔
427	self.logger.info(_("session.new_session"))	1✔
428	self._remove_json()	1✔
429
430	def _json_exists(self):	1✔
431	# todo: make this location configurable
432	home_path = os.path.expanduser("~")	1✔
433	file_path = os.path.join(home_path, "tableau_auth.json")	1✔
434	return os.path.exists(file_path)	1✔
435
436	def _save_session_to_json(self):	1✔
437	try:	1✔
438	data = self._serialize_for_save()	1✔
439	self._save_file(data)	1✔
440	except Exception as e:	×
441	self._wipe_bad_json(e, "Failed to save session file")	×
442
443	def _save_file(self, data):	1✔
444	file_path = self._get_file_path()	1✔
445	with open(str(file_path), "w") as f:	1✔
446	json.dump(data, f)	1✔
447
448	def _serialize_for_save(self):	1✔
449	data = {"tableau_auth": []}	1✔
450	data["tableau_auth"].append(	1✔
451	{
452	"auth_token": self.auth_token,
453	"server": self.server_url,
454	"username": self.username,
455	"user_id": self.user_id,
456	"site_name": self.site_name,
457	"site_id": self.site_id,
458	"personal_access_token_name": self.token_name,
459	"personal_access_token": self.token_value,
460	"last_login_using": self.last_login_using,
461	"password_file": self.password_file,
462	"token_file": self.token_file,
463	"no_prompt": self.no_prompt,
464	"no_certcheck": self.no_certcheck,
465	"certificate": self.certificate,
466	"no_proxy": self.no_proxy,
467	"proxy": self.proxy,
468	"timeout": self.timeout,
469	}
470	)
471	return data	1✔
472
473	def _remove_json(self):	1✔
474	file_path = ""	1✔
475	try:	1✔
476	if not self._json_exists():	1✔
477	return	1✔
478	file_path = self._get_file_path()	1✔
479	self._save_file({})	1✔
480	if os.path.exists(file_path):	1✔
481	os.remove(file_path)	1✔
482	except Exception as e:	1✔
483	message = "Error clearing session data from {}: check and remove manually".format(file_path)	1✔
484	self.logger.error(message)	1✔
485	self.logger.error(e)	1✔
486
487	def _get_server_base_url(self, url: str):	1✔
488	try:	1✔
489	parsed = urlparse(url)	1✔
490	scheme = parsed.scheme or "http"	1✔
491
492	# If netloc is empty, treat path as netloc and discard any extra path
493	netloc = parsed.netloc or parsed.path.split("/")[0] # Keep only the domain	1✔
494	return urlunparse((scheme, netloc, "", "", "", ""))	1✔
NEW 495	except Exception as e:	×
NEW 496	Errors.exit_with_error(self.logger, exception=e)	×

tableau / tabcmd / 13595221517

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous