# -*- coding: utf-8 -*-
#
# Copyright (C) 2013-2017 Gauvain Pocentek
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see .
import configparser
import os
import shlex
import subprocess
from os.path import expanduser, expandvars
from typing import List, Optional, Union
from gitlab.const import USER_AGENT
def _env_config() -> List[str]:
if "PYTHON_GITLAB_CFG" in os.environ:
return [os.environ["PYTHON_GITLAB_CFG"]]
return []
_DEFAULT_FILES: List[str] = _env_config() + [
"/etc/python-gitlab.cfg",
os.path.expanduser("~/.python-gitlab.cfg"),
]
HELPER_PREFIX = "helper:"
HELPER_ATTRIBUTES = ["job_token", "http_password", "private_token", "oauth_token"]
class ConfigError(Exception):
pass
class GitlabIDError(ConfigError):
pass
class GitlabDataError(ConfigError):
pass
class GitlabConfigMissingError(ConfigError):
pass
class GitlabConfigHelperError(ConfigError):
pass
class GitlabConfigParser(object):
def __init__(
self, gitlab_id: Optional[str] = None, config_files: Optional[List[str]] = None
) -> None:
self.gitlab_id = gitlab_id
_files = config_files or _DEFAULT_FILES
file_exist = False
for file in _files:
if os.path.exists(file):
file_exist = True
if not file_exist:
raise GitlabConfigMissingError(
"Config file not found. \nPlease create one in "
"one of the following locations: {} \nor "
"specify a config file using the '-c' parameter.".format(
", ".join(_DEFAULT_FILES)
)
)
self._config = configparser.ConfigParser()
self._config.read(_files)
if self.gitlab_id is None:
try:
self.gitlab_id = self._config.get("global", "default")
except Exception as e:
raise GitlabIDError(
"Impossible to get the gitlab id (not specified in config file)"
) from e
try:
self.url = self._config.get(self.gitlab_id, "url")
except Exception as e:
raise GitlabDataError(
"Impossible to get gitlab informations from "
"configuration (%s)" % self.gitlab_id
) from e
self.ssl_verify: Union[bool, str] = True
try:
self.ssl_verify = self._config.getboolean("global", "ssl_verify")
except ValueError:
# Value Error means the option exists but isn't a boolean.
# Get as a string instead as it should then be a local path to a
# CA bundle.
try:
self.ssl_verify = self._config.get("global", "ssl_verify")
except Exception:
pass
except Exception:
pass
try:
self.ssl_verify = self._config.getboolean(self.gitlab_id, "ssl_verify")
except ValueError:
# Value Error means the option exists but isn't a boolean.
# Get as a string instead as it should then be a local path to a
# CA bundle.
try:
self.ssl_verify = self._config.get(self.gitlab_id, "ssl_verify")
except Exception:
pass
except Exception:
pass
self.timeout = 60
try:
self.timeout = self._config.getint("global", "timeout")
except Exception:
pass
try:
self.timeout = self._config.getint(self.gitlab_id, "timeout")
except Exception:
pass
self.private_token = None
try:
self.private_token = self._config.get(self.gitlab_id, "private_token")
except Exception:
pass
self.oauth_token = None
try:
self.oauth_token = self._config.get(self.gitlab_id, "oauth_token")
except Exception:
pass
self.job_token = None
try:
self.job_token = self._config.get(self.gitlab_id, "job_token")
except Exception:
pass
self.http_username = None
self.http_password = None
try:
self.http_username = self._config.get(self.gitlab_id, "http_username")
self.http_password = self._config.get(self.gitlab_id, "http_password")
except Exception:
pass
self._get_values_from_helper()
self.api_version = "4"
try:
self.api_version = self._config.get("global", "api_version")
except Exception:
pass
try:
self.api_version = self._config.get(self.gitlab_id, "api_version")
except Exception:
pass
if self.api_version not in ("4",):
raise GitlabDataError("Unsupported API version: %s" % self.api_version)
self.per_page = None
for section in ["global", self.gitlab_id]:
try:
self.per_page = self._config.getint(section, "per_page")
except Exception:
pass
if self.per_page is not None and not 0 <= self.per_page <= 100:
raise GitlabDataError("Unsupported per_page number: %s" % self.per_page)
self.pagination = None
try:
self.pagination = self._config.get(self.gitlab_id, "pagination")
except Exception:
pass
self.order_by = None
try:
self.order_by = self._config.get(self.gitlab_id, "order_by")
except Exception:
pass
self.user_agent = USER_AGENT
try:
self.user_agent = self._config.get("global", "user_agent")
except Exception:
pass
try:
self.user_agent = self._config.get(self.gitlab_id, "user_agent")
except Exception:
pass
self.retry_transient_errors = False
try:
self.retry_transient_errors = self._config.getboolean(
"global", "retry_transient_errors"
)
except Exception:
pass
try:
self.retry_transient_errors = self._config.getboolean(
self.gitlab_id, "retry_transient_errors"
)
except Exception:
pass
def _get_values_from_helper(self) -> None:
"""Update attributes that may get values from an external helper program"""
for attr in HELPER_ATTRIBUTES:
value = getattr(self, attr)
if not isinstance(value, str):
continue
if not value.lower().strip().startswith(HELPER_PREFIX):
continue
helper = value[len(HELPER_PREFIX) :].strip()
commmand = [expanduser(expandvars(token)) for token in shlex.split(helper)]
try:
value = (
subprocess.check_output(commmand, stderr=subprocess.PIPE)
.decode("utf-8")
.strip()
)
except subprocess.CalledProcessError as e:
stderr = e.stderr.decode().strip()
raise GitlabConfigHelperError(
f"Failed to read {attr} value from helper "
f"for {self.gitlab_id}:\n{stderr}"
) from e
setattr(self, attr, value)