Source code for saltext.github.states.github

"""
**Deprecated**: Manage organization teams/members, repositories and rulesets.

.. deprecated:: 1.1.0

    Use the :py:mod:`github_org <saltext.github.states.github_org>` and
    :py:mod:`github_repo <saltext.github.states.github_repo>` state modules
    instead.
"""

import datetime
import logging
import time

from salt.exceptions import CommandExecutionError

from saltext.github.utils.gh.statecores import catch_errors as _catch_errors

log = logging.getLogger(__name__)

__virtualname__ = "github"


[docs] def __virtual__(): """ Only load if the github module is available in __salt__ """ if "github.list_users" in __salt__: return __virtualname__ return (False, "github module could not be loaded") # pragma: no cover
[docs] @_catch_errors def present(name, profile="github", **kwargs): """ Ensure a user is present .. deprecated:: 1.1.0 Use :py:func:`github_org.member_present <saltext.github.states.github_org.member_present>` instead. .. code-block:: yaml ensure user test is present in github: github.present: - name: gitexample name This is the github handle of the user in the organization """ ret = {"name": name, "changes": {}, "result": None, "comment": ""} target = __salt__["github.get_user"](name, profile=profile, **kwargs) # If the user has a valid github handle and is not in the org already if not target: ret["result"] = False ret["comment"] = f"Couldnt find user {name}" elif isinstance(target, bool) and target: ret["comment"] = f"User {name} is already in the org " ret["result"] = True elif not target.get("in_org", False) and target.get("membership_state") != "pending": if __opts__["test"]: ret["comment"] = f"User {name} will be added to the org" ret["changes"]["old"] = None ret["changes"]["new"] = f"User {name} exists in the org now" return ret # add the user result = __salt__["github.add_user"](name, profile=profile, **kwargs) if result: ret["changes"].setdefault("old", None) ret["changes"].setdefault("new", f"User {name} exists in the org now") ret["result"] = True else: ret["result"] = False ret["comment"] = f"Failed to add user {name} to the org" else: ret["comment"] = f"User {name} has already been invited." ret["result"] = True return ret
[docs] @_catch_errors def absent(name, profile="github", **kwargs): """ Ensure a github user is absent .. deprecated:: 1.1.0 Use :py:func:`github_org.member_absent <saltext.github.states.github_org.member_absent>` instead. .. code-block:: yaml ensure user test is absent in github: github.absent: - name: Example TestUser1 - email: example@domain.com - username: gitexample name Github handle of the user in organization """ ret = { "name": name, "changes": {}, "result": None, "comment": f"User {name} is absent.", } target = __salt__["github.get_user"](name, profile=profile, **kwargs) if target: if isinstance(target, bool) or target.get("in_org", False): if __opts__["test"]: ret["comment"] = f"User {name} will be deleted" ret["changes"]["old"] = f"User {name} exists" ret["changes"]["new"] = f"User {name} deleted" ret["result"] = None return ret result = __salt__["github.remove_user"](name, profile=profile, **kwargs) if result: ret["comment"] = f"Deleted user {name}" ret["changes"].setdefault("old", f"User {name} exists") ret["changes"].setdefault("new", f"User {name} deleted") ret["result"] = True else: ret["comment"] = f"Failed to delete {name}" ret["result"] = False else: ret["comment"] = f"User {name} has already been deleted!" ret["result"] = True else: ret["comment"] = f"User {name} does not exist" ret["result"] = True return ret return ret
[docs] @_catch_errors def team_present( # pylint: disable=too-many-statements name, description=None, repo_names=None, privacy="secret", permission="pull", members=None, enforce_mfa=False, no_mfa_grace_seconds=0, profile="github", **kwargs, ): """ Ensure a team is present .. deprecated:: 1.1.0 Use :py:func:`github_org.team_present <saltext.github.states.github_org.team_present>` instead. name This is the name of the team in the organization. description The description of the team. repo_names The names of repositories to add the team to. privacy The level of privacy for the team, can be ``secret`` or ``closed``. Defaults to secret. permission The default permission for new repositories added to the team, can be ``pull``, ``push`` or ``admin``. Defaults to pull. members The members belonging to the team, specified as a dict of member name to optional configuration. Options include ``enforce_mfa_from`` and ``mfa_exempt``. enforce_mfa Whether to enforce MFA requirements on members of the team. If True then all members without ``mfa_exempt: True`` configured will be removed from the team. Note that ``no_mfa_grace_seconds`` may be set to allow members a grace period. no_mfa_grace_seconds The number of seconds of grace time that a member will have to enable MFA before being removed from the team. The grace period will begin from ``enforce_mfa_from`` on the member configuration, which defaults to 1970/01/01. Example: .. code-block:: yaml Ensure team test is present in github: github.team_present: - name: test - members: user1: {} user2: {} Ensure team test_mfa is present in github: github.team_present: - name: test_mfa - members: user1: enforce_mfa_from: 2016/06/15 - enforce_mfa: True """ ret = {"name": name, "changes": {}, "result": True, "comment": ""} target = __salt__["github.get_team"](name, profile=profile, **kwargs) test_comments = [] if target: # Team already exists parameters = {} if description is not None and target["description"] != description: parameters["description"] = description if permission is not None and target["permission"] != permission: parameters["permission"] = permission if privacy is not None and target["privacy"] != privacy: parameters["privacy"] = privacy if len(parameters) > 0: team_change = { "old": f"Team properties were {target}", "new": f"Team properties (that changed) are {parameters}", } if __opts__["test"]: test_comments.append(f"Team properties are set to be edited: {parameters}") ret["result"] = None ret["changes"]["team"] = team_change else: result = __salt__["github.edit_team"](name, profile=profile, **parameters) if result: ret["changes"]["team"] = team_change else: ret["result"] = False ret["comment"] = "Failed to update team properties." return ret current_repos = set(__salt__["github.list_team_repos"](name, profile=profile)) repo_names = set(repo_names or []) repos_to_add = repo_names - current_repos repos_to_remove = current_repos - repo_names if repo_names else [] if repos_to_add: if __opts__["test"]: test_comments.append( "Team {} will have the following repos added: {}.".format( # pylint: disable=consider-using-f-string name, list(repos_to_add) ) ) ret["result"] = None for repo_name in repos_to_add: repo_change = { "old": f"Repo {repo_name} is not in team {name}", "new": f"Repo {repo_name} is in team {name}", } if __opts__["test"]: ret["changes"][repo_name] = repo_change else: result = __salt__["github.add_team_repo"]( repo_name, name, profile=profile, **kwargs ) if result: ret["changes"][repo_name] = repo_change else: ret["result"] = False ret["comment"] = f"Failed to add repo {repo_name} to team {name}." return ret if repos_to_remove: if __opts__["test"]: test_comments.append( "Team {} will have the following repos removed: {}.".format( # pylint: disable=consider-using-f-string name, list(repos_to_remove) ) ) ret["result"] = None for repo_name in repos_to_remove: repo_change = { "old": f"Repo {repo_name} is in team {name}", "new": f"Repo {repo_name} is not in team {name}", } if __opts__["test"]: ret["changes"][repo_name] = repo_change else: result = __salt__["github.remove_team_repo"]( repo_name, name, profile=profile, **kwargs ) if result: ret["changes"][repo_name] = repo_change else: ret["result"] = False ret["comment"] = f"Failed to remove repo {repo_name} from team {name}." return ret else: # Team does not exist - it will be created. if __opts__["test"]: # Note: member/repo changes are not projected for new teams ret["comment"] = f"Team {name} is set to be created." ret["changes"]["team"] = {"old": None, "new": f"Team {name} has been created"} ret["result"] = None return ret result = __salt__["github.add_team"]( name, description=description, repo_names=repo_names, permission=permission, privacy=privacy, profile=profile, **kwargs, ) if result: ret["changes"]["team"] = {} ret["changes"]["team"]["old"] = None ret["changes"]["team"]["new"] = f"Team {name} has been created" else: ret["result"] = False ret["comment"] = f"Failed to create team {name}." return ret manage_members = members is not None mfa_deadline = datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta( seconds=no_mfa_grace_seconds ) members_no_mfa = __salt__["github.list_members_without_mfa"](profile=profile) members_lower = {} for member_name, info in (members or {}).items(): members_lower[member_name.lower()] = info member_change = False current_members = __salt__["github.list_team_members"](name, profile=profile) for member, member_info in (members or {}).items(): log.info("Checking member %s in team %s", member, name) if member.lower() not in current_members: if enforce_mfa and _member_violates_mfa( member, member_info, mfa_deadline, members_no_mfa ): if __opts__["test"]: test_comments.append( f"User {member} will not be added to the team because they do not have MFA." ) else: # Add to team member_change = True member_add = { "old": f"User {member} is not in team {name}", "new": f"User {member} is in team {name}", } if __opts__["test"]: test_comments.append(f"User {member} set to be added to the team.") ret["result"] = None ret["changes"][member] = member_add else: result = __salt__["github.add_team_member"]( member, name, profile=profile, **kwargs ) if result: ret["changes"][member] = member_add else: ret["result"] = False ret["comment"] = f"Failed to add user {member} to team {name}." return ret for member in current_members: mfa_violation = False if member in members_lower: mfa_violation = _member_violates_mfa( member, members_lower[member], mfa_deadline, members_no_mfa ) if manage_members and member not in members_lower or (enforce_mfa and mfa_violation): # Remove from team member_change = True extra_changes = " due to MFA violation" if mfa_violation else "" member_removal = { "old": f"User {member} is in team {name}", "new": f"User {member} is not in team {name}{extra_changes}", } if __opts__["test"]: if mfa_violation: test_comments.append( f"User {member} set to be removed from the team because they do not have MFA." ) else: test_comments.append(f"User {member} set to be removed from the team.") ret["result"] = None ret["changes"][member] = member_removal else: result = __salt__["github.remove_team_member"]( member, name, profile=profile, **kwargs ) if result: ret["changes"][member] = member_removal else: ret["result"] = False ret["comment"] = f"Failed to remove user {member} from team {name}." return ret if member_change: # Refresh team cache __salt__["github.list_team_members"](name, profile=profile, ignore_cache=False, **kwargs) if len(test_comments) > 0: ret["comment"] = "\n".join(test_comments) return ret
def _member_violates_mfa(member, member_info, mfa_deadline, members_without_mfa): if member_info.get("mfa_exempt", False): return False enforce_mfa_from = datetime.datetime.strptime( member_info.get("enforce_mfa_from", "1970/01/01"), "%Y/%m/%d" ).replace(tzinfo=datetime.timezone.utc) return member.lower() in members_without_mfa and (mfa_deadline > enforce_mfa_from)
[docs] @_catch_errors def team_absent(name, profile="github", **kwargs): """ Ensure a team is absent. .. deprecated:: 1.1.0 Use :py:func:`github_org.team_absent <saltext.github.states.github_org.team_absent>` instead. Example: .. code-block:: yaml ensure team test is present in github: github.team_absent: - name: test name This is the name of the team in the organization. """ ret = {"name": name, "changes": {}, "result": None, "comment": ""} target = __salt__["github.get_team"](name, profile=profile, **kwargs) if not target: ret["comment"] = f"Team {name} does not exist" ret["result"] = True return ret if __opts__["test"]: ret["comment"] = f"Team {name} will be deleted" ret["changes"]["old"] = f"Team {name} exists" ret["changes"]["new"] = f"Team {name} deleted" ret["result"] = None return ret result = __salt__["github.remove_team"](name, profile=profile, **kwargs) if result: ret["comment"] = f"Deleted team {name}" ret["changes"].setdefault("old", f"Team {name} exists") ret["changes"].setdefault("new", f"Team {name} deleted") ret["result"] = True else: ret["comment"] = f"Failed to delete {name}" ret["result"] = False return ret
[docs] @_catch_errors def repo_present( name, description=None, homepage=None, private=None, has_issues=None, has_wiki=None, has_downloads=None, auto_init=False, gitignore_template=None, license_template=None, teams=None, profile="github", **kwargs, ): """ Ensure a repository is present .. deprecated:: 1.1.0 Use :py:func:`github_repo.present <saltext.github.states.github_repo.present>` instead. name This is the name of the repository. description The description of the repository. homepage The URL with more information about the repository. private The visiblity of the repository. Note that private repositories require a paid GitHub account. has_issues Whether to enable issues for this repository. has_wiki Whether to enable the wiki for this repository. has_downloads Whether to enable downloads for this repository. auto_init Whether to create an initial commit with an empty README. gitignore_template The desired language or platform for a .gitignore, e.g "Haskell". license_template The desired LICENSE template to apply, e.g "mit" or "mozilla". teams The teams for which this repo should belong to, specified as a dict of team name to permission (``pull``, ``push`` or ``admin``). Example: .. code-block:: yaml Ensure repo my-repo is present in github: github.repo_present: - name: my-repo - description: My very important repository """ ret = {"name": name, "changes": {}, "result": True, "comment": ""} # This is an optimization to cache all repos in the organization up front. # The first use of this state will collect all of the repos and save a bunch # of API calls for future use. __salt__["github.list_repos"](profile=profile) try: target = __salt__["github.get_repo_info"](name, profile=profile, **kwargs) except CommandExecutionError: target = None given_params = { "description": description, "homepage": homepage, "private": private, "has_issues": has_issues, "has_wiki": has_wiki, "has_downloads": has_downloads, "auto_init": auto_init, "gitignore_template": gitignore_template, "license_template": license_template, } # Keep track of current_teams if we've fetched them after creating a new repo current_teams = None if target: # Repo already exists # Some params are only valid on repo creation ignore_params = ["auto_init", "gitignore_template", "license_template"] parameters = {} old_parameters = {} for param_name, param_value in given_params.items(): if ( param_value is not None and param_name not in ignore_params and target[param_name] is not param_value and target[param_name] != param_value ): parameters[param_name] = param_value old_parameters[param_name] = target[param_name] if len(parameters) > 0: repo_change = { "old": f"Repo properties were {old_parameters}", "new": f"Repo properties (that changed) are {parameters}", } if __opts__["test"]: ret["changes"]["repo"] = repo_change ret["result"] = None else: result = __salt__["github.edit_repo"](name, profile=profile, **parameters) if result: ret["changes"]["repo"] = repo_change else: ret["result"] = False ret["comment"] = "Failed to update repo properties." return ret else: # Repo does not exist - it will be created. repo_change = {"old": None, "new": f"Repo {name} has been created"} if __opts__["test"]: ret["changes"]["repo"] = repo_change ret["result"] = None else: add_params = dict(given_params) add_params.update(kwargs) result = __salt__["github.add_repo"](name, profile=profile, **add_params) if not result: ret["result"] = False ret["comment"] = f"Failed to create repo {name}." return ret # Turns out that trying to fetch teams for a new repo can 404 immediately # after repo creation, so this waits until we can fetch teams successfully # before continuing. for attempt in range(3): time.sleep(1) try: current_teams = __salt__["github.get_repo_teams"]( name, profile=profile, **kwargs ) break except CommandExecutionError: log.info("Attempt %s to fetch new repo %s failed", attempt, name) if current_teams is None: ret["result"] = False ret["comment"] = f"Failed to verify repo {name} after creation." return ret ret["changes"]["repo"] = repo_change if teams is not None: if __opts__["test"] and not target: # Assume no teams if we're in test mode and the repo doesn't exist current_teams = [] elif current_teams is None: current_teams = __salt__["github.get_repo_teams"](name, profile=profile) current_team_names = {t["name"] for t in current_teams} # First remove any teams that aren't present for team_name in current_team_names: if team_name in teams: continue team_change = { "old": f"Repo {name} is in team {team_name}", "new": f"Repo {name} is not in team {team_name}", } if __opts__["test"]: ret["changes"][team_name] = team_change ret["result"] = None else: result = __salt__["github.remove_team_repo"](name, team_name, profile=profile) if not result: ret["result"] = False ret["comment"] = f"Failed to remove repo {name} from team {team_name}." return ret ret["changes"][team_name] = team_change # Next add or modify any necessary teams for team_name, permission in teams.items(): if team_name not in current_team_names: # Need to add repo to team team_change = { "old": f"Repo {name} is not in team {team_name}", "new": f"Repo {name} is in team {team_name}", } if __opts__["test"]: ret["changes"][team_name] = team_change ret["result"] = None else: result = __salt__["github.add_team_repo"]( name, team_name, profile=profile, permission=permission ) if not result: ret["result"] = False ret["comment"] = f"Failed to add repo {name} to team {team_name}." return ret ret["changes"][team_name] = team_change else: current_permission = ( __salt__["github.list_team_repos"](team_name, profile=profile) .get(name.lower(), {}) .get("permission") ) if not current_permission: ret["result"] = False ret["comment"] = ( f"Failed to determine current permission for team {team_name} in repo {name}" ) return ret if current_permission == permission: continue team_change = { "old": f"Repo {name} in team {team_name} has permission {current_permission}", "new": f"Repo {name} in team {team_name} has permission {permission}", } if __opts__["test"]: ret["changes"][team_name] = team_change ret["result"] = None else: result = __salt__["github.add_team_repo"]( name, team_name, profile=profile, permission=permission ) if not result: ret["result"] = False ret["comment"] = ( f"Failed to set permission on repo {name} from team {team_name} to {permission}." ) return ret ret["changes"][team_name] = team_change return ret
[docs] @_catch_errors def repo_absent(name, profile="github", **kwargs): """ Ensure a repo is absent. .. deprecated:: 1.1.0 Use :py:func:`github_repo.absent <saltext.github.states.github_repo.absent>` instead. Example: .. code-block:: yaml ensure repo test is absent in github: github.repo_absent: - name: test name This is the name of the repository in the organization. """ ret = {"name": name, "changes": {}, "result": None, "comment": ""} try: target = __salt__["github.get_repo_info"](name, profile=profile, **kwargs) except CommandExecutionError: target = None if not target: ret["comment"] = f"Repo {name} does not exist" ret["result"] = True return ret if __opts__["test"]: ret["comment"] = f"Repo {name} will be deleted" ret["changes"]["old"] = f"Repo {name} exists" ret["changes"]["new"] = f"Repo {name} deleted" ret["result"] = None return ret result = __salt__["github.remove_repo"](name, profile=profile, **kwargs) if result: ret["comment"] = f"Deleted repo {name}" ret["changes"].setdefault("old", f"Repo {name} exists") ret["changes"].setdefault("new", f"Repo {name} deleted") ret["result"] = True return ret ret["comment"] = ( f"Failed to delete repo {name}. Ensure the delete_repo scope is enabled if using OAuth." ) ret["result"] = False return ret
[docs] @_catch_errors def ruleset_absent( name, ruleset_type, profile="github", owner=None, repo_name=None, org_name=None, **kwargs ): """ Ensure a ruleset is absent. .. deprecated:: 1.1.0 Use :py:func:`github_org.ruleset_absent <saltext.github.states.github_org.ruleset_absent>` or :py:func:`github_repo.ruleset_absent <saltext.github.states.github_repo.ruleset_absent>` instead. Example: .. code-block:: yaml ensure rulset test is absent in github: github.ruleset_absent: - name: test - ruleset_type: repo - owner: test - repo_name: test_repo name name of the ruleset. ruleset_type type of ruleset (``org`` or ``repo``) """ ret = {"name": name, "changes": {}, "result": None, "comment": ""} if ruleset_type not in ["org", "repo"]: ret["result"] = False ret["comment"] = "ruleset_type not set to repo or org" return ret if ruleset_type == "repo": kwargs.update({"owner": owner, "repo_name": repo_name, "ruleset_type": ruleset_type}) if ruleset_type == "org": kwargs.update({"org_name": org_name, "ruleset_type": ruleset_type}) rulesets = __salt__["github.list_rulesets"](profile, **kwargs) if rulesets.get("result") is False: ret["result"] = False ret["comment"] = "error listing rulesets" return ret ruleset = next( (ruleset for ruleset in (rulesets.get("rulesets") or []) if ruleset["name"] == name), None ) if ruleset is None: ret["comment"] = f"Ruleset {name} does not exist" ret["result"] = True return ret if __opts__["test"]: ret["comment"] = f"Ruleset {name} will be deleted" ret["changes"]["old"] = f"ruleset {name} exists" ret["changes"]["new"] = f"ruleset {name} deleted" ret["result"] = None return ret kwargs.update({"ruleset_id": ruleset["id"]}) result = __salt__["github.delete_ruleset"](profile, **kwargs) # delete_ruleset returns a (truthy) dict with result: False on failure if result.get("result") is not False: ret["comment"] = f"Deleted ruleset {name}" ret["changes"]["old"] = f"ruleset {name} exists" ret["changes"]["new"] = f"ruleset {name} deleted" ret["result"] = True return ret ret["comment"] = f"Failed to delete ruleset {name}" ret["result"] = False return ret
[docs] @_catch_errors def ruleset_present( name, ruleset_type, profile="github", owner=None, repo_name=None, org_name=None, ruleset_params=None, **kwargs, ): """ Ensure a ruleset is present. .. deprecated:: 1.1.0 Use :py:func:`github_org.ruleset_present <saltext.github.states.github_org.ruleset_present>` or :py:func:`github_repo.ruleset_present <saltext.github.states.github_repo.ruleset_present>` instead. Example: .. code-block:: yaml ensure rulset test is present in github: github.ruleset_present: - name: test - ruleset_type: repo - owner: test - repo_name: test_repo - ruleset_params: target: branch enforcement: disabled name name of the ruleset. ruleset_type type of ruleset (``org`` or ``repo``) ruleset_params parameters to set the ruleset rules. i.e ``{enforcement: disabled}`` """ ret = {"name": name, "changes": {}, "result": True, "comment": ""} if ruleset_type not in ["org", "repo"]: ret["result"] = False ret["comment"] = "ruleset_type not set to repo or org" return ret if ruleset_type == "repo": kwargs.update({"owner": owner, "repo_name": repo_name, "ruleset_type": ruleset_type}) if ruleset_type == "org": kwargs.update({"org_name": org_name, "ruleset_type": ruleset_type}) if ruleset_params: ruleset_params["name"] = name else: ruleset_params = {} rulesets = __salt__["github.list_rulesets"](profile, **kwargs) if rulesets.get("result") is False: ret["result"] = False ret["comment"] = "error listing rulesets" return ret ruleset = next( (ruleset for ruleset in (rulesets.get("rulesets") or []) if ruleset["name"] == name), None ) if ruleset is None: ret["changes"] = {"old": "No existing ruleset found", "new": "Ruleset created"} if __opts__["test"]: ret["result"] = None ret["comment"] = "ruleset will be added" return ret result = __salt__["github.add_ruleset"](profile, ruleset_params=ruleset_params, **kwargs) ret["comment"] = "ruleset added" if not result.get("id"): ret["changes"] = {} ret["result"] = False ret["comment"] = "ruleset could not be added" return ret kwargs.update({"ruleset_id": ruleset["id"]}) ruleset_info = __salt__["github.get_ruleset"](profile, **kwargs) if not ruleset_info.get("id"): ret["comment"] = "Could not get ruleset" ret["result"] = False return ret # Only diff the parameters that are being managed. The server # response carries additional server-populated fields (id, # timestamps, source, ...) that must not register as drift. diff = {key: value for key, value in ruleset_params.items() if ruleset_info.get(key) != value} if not diff: ret["comment"] = "ruleset present" return ret ret["changes"] = { "old": {key: ruleset_info.get(key) for key in diff}, "new": diff, } if __opts__["test"]: ret["result"] = None ret["comment"] = "ruleset will be updated" return ret kwargs.update({"ruleset_params": ruleset_params}) result = __salt__["github.update_ruleset"](profile, **kwargs) if result.get("id"): ret["comment"] = "ruleset updated" else: ret["changes"] = {} ret["result"] = False ret["comment"] = "Could not update ruleset" return ret