github_org_control: Update to the latest version (#9322)
This commit is contained in:
Alexander Zhogov authored and 
GitHub committed
1 parent
46c9755b6e
commit
c10a27f601
15 files changed
+672
-499
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,139 @@
|
||||
# Copyright (C) 2018-2021 Intel Corporation
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
"""
|
||||
Check GitHub organization and invite members
|
||||
"""
|
||||
|
||||
# pylint: disable=fixme,no-member,too-many-locals
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from argparse import ArgumentParser
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parents[1]))
|
||||
from github_org_control.configs import Config
|
||||
from github_org_control.github_api import GithubOrgApi, get_dev_emails, print_users
|
||||
from github_org_control.ldap_api import LdapApi, print_user_info, InfoLevel
|
||||
|
||||
|
||||
def remove_members(gh_api, cfg_emails, org_emails, dev_emails, org_emails_no_in_ldap):
|
||||
"""Checks and remove members"""
|
||||
print(
|
||||
f"\n{'=' * 10} Check accounts below and remove from the GitHub organization or "
|
||||
f"configuration {'=' * 10}"
|
||||
)
|
||||
|
||||
cfg_emails_no_in_org = sorted(cfg_emails.difference(org_emails))
|
||||
print(
|
||||
f"\nCfg developer emails - absent in GitHub organization {len(cfg_emails_no_in_org)}:",
|
||||
"; ".join(cfg_emails_no_in_org),
|
||||
)
|
||||
|
||||
non_member_ignored_logins = set(Config().IGNORE_LOGINS).difference(
|
||||
set(gh_api.org_members_by_login.keys())
|
||||
)
|
||||
print(
|
||||
f"\nIgnored logins - absent in GitHub organization {len(non_member_ignored_logins)}:\n",
|
||||
"\n".join(non_member_ignored_logins),
|
||||
)
|
||||
|
||||
org_emails_no_in_dev = sorted(org_emails.difference(dev_emails))
|
||||
print(
|
||||
f"\nOrg member emails - absent in cfg and LDAP PDLs {len(org_emails_no_in_dev)}:",
|
||||
"; ".join(org_emails_no_in_dev),
|
||||
)
|
||||
|
||||
print(
|
||||
f"\nOrg member emails - absent in LDAP at all {len(org_emails_no_in_ldap)}:",
|
||||
"; ".join(sorted(org_emails_no_in_ldap)),
|
||||
)
|
||||
|
||||
print("\nOrg members - no real name:")
|
||||
members_to_fix_name = sorted(gh_api.members_to_fix_name, key=lambda member: member.email)
|
||||
print_users(members_to_fix_name)
|
||||
print(
|
||||
"\nOrg member emails - no real name:",
|
||||
"; ".join([member.email.lower() for member in members_to_fix_name]),
|
||||
)
|
||||
|
||||
print("\nOrg members - no Intel emails:")
|
||||
print_users(gh_api.members_to_remove)
|
||||
|
||||
gh_api.remove_users(org_emails_no_in_ldap | gh_api.members_to_remove)
|
||||
|
||||
|
||||
def main():
|
||||
"""The main entry point function"""
|
||||
arg_parser = ArgumentParser()
|
||||
arg_parser.add_argument(
|
||||
"--cfg-file",
|
||||
metavar="PATH",
|
||||
default=Config.default_cfg_path,
|
||||
help=f"Path to json configuration file, e.g. {Config.default_cfg_path}",
|
||||
)
|
||||
arg_parser.add_argument("--teams", action="store_true", help="Check GitHub teams")
|
||||
arg_parser.add_argument("--no-ldap", action="store_true", help="Don't use LDAP info")
|
||||
args, unknown_args = arg_parser.parse_known_args()
|
||||
|
||||
Config(args.cfg_file, unknown_args)
|
||||
gh_api = GithubOrgApi()
|
||||
|
||||
if args.teams:
|
||||
gh_api.get_org_teams()
|
||||
return
|
||||
|
||||
cfg_emails = get_dev_emails()
|
||||
print(f"\nCfg developer emails {len(cfg_emails)}:", "; ".join(sorted(cfg_emails)))
|
||||
|
||||
dev_emails = set()
|
||||
dev_emails.update(cfg_emails)
|
||||
|
||||
if not args.no_ldap:
|
||||
ldap_api = LdapApi()
|
||||
ldap_emails = ldap_api.get_user_emails()
|
||||
dev_emails.update(ldap_emails)
|
||||
print(f"\nLDAP developer emails {len(ldap_emails)}:", "; ".join(sorted(ldap_emails)))
|
||||
|
||||
cfg_emails_no_in_ldap = ldap_api.get_absent_emails(cfg_emails)
|
||||
print(
|
||||
f"\nCfg developer emails - absent in LDAP at all {len(cfg_emails_no_in_ldap)}:",
|
||||
"; ".join(sorted(cfg_emails_no_in_ldap)),
|
||||
)
|
||||
|
||||
cfg_ldap_inters = cfg_emails.intersection(ldap_emails)
|
||||
print(
|
||||
f"\nCfg developer emails - present in LDAP developers {len(cfg_ldap_inters)}:",
|
||||
"; ".join(sorted(cfg_ldap_inters)),
|
||||
)
|
||||
|
||||
org_emails = gh_api.get_org_emails()
|
||||
print(f"\nOrg emails {len(org_emails)}:", "; ".join(sorted(org_emails)))
|
||||
|
||||
org_emails_no_in_ldap = set()
|
||||
if not args.no_ldap:
|
||||
org_ldap_diff = org_emails.difference(ldap_emails)
|
||||
print(
|
||||
f"\nOrg member emails - absent in LDAP developers {len(org_ldap_diff)}:",
|
||||
"; ".join(sorted(org_ldap_diff)),
|
||||
)
|
||||
|
||||
for email in org_ldap_diff:
|
||||
user_info = ldap_api.get_user_info_by_email(email)
|
||||
if user_info:
|
||||
print_user_info(user_info, InfoLevel.PDL)
|
||||
else:
|
||||
org_emails_no_in_ldap.add(email)
|
||||
|
||||
org_pendig_invitation_emails = gh_api.get_org_invitation_emails()
|
||||
invite_emails = dev_emails.difference(org_emails).difference(org_pendig_invitation_emails)
|
||||
print(f"\nInvite emails {len(invite_emails)}:", "; ".join(sorted(invite_emails)))
|
||||
|
||||
valid_github_users = gh_api.get_valid_github_users(invite_emails)
|
||||
gh_api.invite_users(valid_github_users)
|
||||
|
||||
remove_members(gh_api, cfg_emails, org_emails, dev_emails, org_emails_no_in_ldap)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,261 @@
|
||||
# Copyright (C) 2018-2021 Intel Corporation
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
"""
|
||||
Check GitHub PRs and set labels by type and categories, e.g. 'ExternalPR', 'category: ci'
|
||||
"""
|
||||
|
||||
# pylint: disable=fixme,no-member
|
||||
|
||||
import re
|
||||
import sys
|
||||
import datetime
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from argparse import ArgumentParser
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parents[1]))
|
||||
from github_org_control import github_api
|
||||
from github_org_control.configs import Config
|
||||
|
||||
|
||||
class PrType(Enum):
|
||||
"""Constants for type of GitHub pull request by author membership"""
|
||||
|
||||
EXTERNAL = "ExternalPR"
|
||||
INTEL = "ExternalIntelPR"
|
||||
ORG = "OpenvinoPR"
|
||||
BAD = "BadPR"
|
||||
|
||||
|
||||
def get_pr_labels(pull):
|
||||
"""Gets PR labels as set"""
|
||||
pr_lables = set()
|
||||
for label in pull.labels:
|
||||
pr_lables.add(label.name)
|
||||
return pr_lables
|
||||
|
||||
|
||||
def set_pr_labels(pull, labels):
|
||||
"""Sets new PR labels (all previously set labels are removed)"""
|
||||
if not labels or Config().DRY_RUN:
|
||||
return
|
||||
print("Set PR labels:", labels)
|
||||
# set_labels() should accept list but fails with empty "AssertionError:"
|
||||
pull.set_labels(labels)
|
||||
|
||||
|
||||
def add_pr_labels(pull, labels):
|
||||
"""Adds PR labels"""
|
||||
if not labels or Config().DRY_RUN:
|
||||
return
|
||||
print("Add PR labels:", labels)
|
||||
for label in labels:
|
||||
pull.add_to_labels(label)
|
||||
|
||||
|
||||
def get_pr_type_by_labels(pull):
|
||||
"""Gets PR type using labels"""
|
||||
pr_lables = get_pr_labels(pull)
|
||||
pr_types = set(type.value for type in PrType)
|
||||
pr_types_labels = pr_lables & pr_types
|
||||
if not pr_types_labels:
|
||||
return None
|
||||
if len(pr_types_labels) > 1:
|
||||
print(f"Duplicated labels: {pr_types_labels}")
|
||||
return PrType.BAD
|
||||
return PrType(PrType(pr_types_labels.pop()))
|
||||
|
||||
|
||||
def get_label_by_team_name_re(team_name):
|
||||
"""Generates label by PR reviwer team name using regular expressions"""
|
||||
if "admins" in team_name:
|
||||
return "category: ci"
|
||||
re_compile_label = re.compile(rf"{Config().GITHUB_REPO}-(.+)-maintainers")
|
||||
re_label = re_compile_label.match(team_name)
|
||||
if re_label:
|
||||
return f"category: {re_label.group(1).strip()}"
|
||||
return None
|
||||
|
||||
|
||||
def get_label_by_team_name_map(team_name):
|
||||
"""Generates label by PR reviwer team name using config map"""
|
||||
return Config().TEAM_TO_LABEL.get(team_name)
|
||||
|
||||
|
||||
def get_category_labels(pull):
|
||||
"""Gets list of category labels by all PR reviwer teams"""
|
||||
labels = []
|
||||
pr_lables = get_pr_labels(pull)
|
||||
for reviewer_team in pull.get_review_requests()[1]:
|
||||
reviewer_label = get_label_by_team_name_map(reviewer_team.name)
|
||||
if reviewer_label and reviewer_label not in pr_lables:
|
||||
labels.append(reviewer_label)
|
||||
return labels
|
||||
|
||||
|
||||
def get_pr_info_str(pull):
|
||||
"""Gets info about PR using a few workarounds"""
|
||||
pr_title = pull.title.encode("ASCII", "ignore").decode()
|
||||
|
||||
# Workaround for PyGithub issue: https://github.com/PyGithub/PyGithub/issues/512
|
||||
pr_created_at = pull.created_at.replace(tzinfo=datetime.timezone.utc).astimezone()
|
||||
|
||||
return (
|
||||
f"PR: {pull.number} - {pr_title} - Created: {pr_created_at} - "
|
||||
f"Labels: {get_pr_labels(pull)} - Type: {get_pr_type_by_labels(pull)}"
|
||||
)
|
||||
|
||||
|
||||
def update_labels(gh_api, pull, non_org_intel_pr_users, non_org_pr_users):
|
||||
"""Checks and updates labels"""
|
||||
print("Check and update labels:")
|
||||
pr_type_by_labels = get_pr_type_by_labels(pull)
|
||||
add_labels = []
|
||||
|
||||
# Checks PR source type
|
||||
if gh_api.is_org_user(pull.user):
|
||||
print(" - Org user")
|
||||
elif github_api.is_intel_email(pull.user.email) or github_api.is_intel_company(
|
||||
pull.user.company
|
||||
):
|
||||
print(" - Non org user with Intel email or company")
|
||||
non_org_intel_pr_users.add(pull.user)
|
||||
if pr_type_by_labels is not PrType.INTEL:
|
||||
print(f'NO "{PrType.INTEL.value}" label: ', end="")
|
||||
github_api.print_users(pull.user)
|
||||
add_labels.append(PrType.INTEL.value)
|
||||
elif github_api.is_user_ignored(pull.user):
|
||||
print(" - IGNORED non org user with NO Intel email or company")
|
||||
else:
|
||||
print(" - Non org user with NO Intel email or company")
|
||||
non_org_pr_users.add(pull.user)
|
||||
if pr_type_by_labels is not PrType.EXTERNAL:
|
||||
print(f'NO "{PrType.EXTERNAL.value}" label: ', end="")
|
||||
github_api.print_users(pull.user)
|
||||
add_labels.append(PrType.EXTERNAL.value)
|
||||
|
||||
add_labels += get_category_labels(pull)
|
||||
add_pr_labels(pull, add_labels)
|
||||
|
||||
|
||||
def get_wrong_commits(pull):
|
||||
"""Returns commits with incorrect user and email"""
|
||||
pr_author_email = (pull.user.email or "").lower()
|
||||
print("GitHub PR author email:", pr_author_email)
|
||||
print("Check commits:")
|
||||
wrong_commits = set()
|
||||
for commit in pull.get_commits():
|
||||
# import pprint; pprint.pprint(commit.raw_data)
|
||||
print("Commit SHA:", commit.sha)
|
||||
# Use raw data because commit author can be non GitHub user
|
||||
commit_author_email = (commit.raw_data["commit"]["author"]["email"] or "").lower()
|
||||
commit_committer_email = (commit.raw_data["commit"]["committer"]["email"] or "").lower()
|
||||
print(" Commit author email:", commit_author_email)
|
||||
print(" Commit committer email:", commit_committer_email)
|
||||
if not github_api.is_valid_user(commit.author):
|
||||
print(
|
||||
" ERROR: User with the commit author email is absent in GitHub:",
|
||||
commit.raw_data["commit"]["author"]["name"],
|
||||
)
|
||||
wrong_commits.add(commit.sha)
|
||||
if not github_api.is_valid_user(commit.committer):
|
||||
print(
|
||||
" ERROR: User with the commit committer email is absent in GitHub:",
|
||||
commit.raw_data["commit"]["committer"]["name"],
|
||||
)
|
||||
wrong_commits.add(commit.sha)
|
||||
if not commit.raw_data["commit"]["verification"]["verified"]:
|
||||
print(
|
||||
" WARNING: The commit is not verified. Reason:",
|
||||
commit.raw_data["commit"]["verification"]["reason"],
|
||||
)
|
||||
if pr_author_email != commit_author_email or pr_author_email != commit_committer_email:
|
||||
print(" WARNING: Commit emails and GitHub PR author public email are differnt")
|
||||
return wrong_commits
|
||||
|
||||
|
||||
def main():
|
||||
"""The main entry point function"""
|
||||
arg_parser = ArgumentParser()
|
||||
arg_parser.add_argument(
|
||||
"--cfg-file",
|
||||
metavar="PATH",
|
||||
default=Config.default_cfg_path,
|
||||
help=f"Path to json configuration file, e.g. {Config.default_cfg_path}",
|
||||
)
|
||||
arg_parser.add_argument(
|
||||
"--pr", metavar="NUMBER", help="Get GitHub pull request with the number"
|
||||
)
|
||||
arg_parser.add_argument(
|
||||
"--pr-state",
|
||||
default="open",
|
||||
choices=["open", "closed"],
|
||||
help="Set GitHub pull request state",
|
||||
)
|
||||
arg_parser.add_argument(
|
||||
"--newer", metavar="MINUTES", help="Get newly created GitHub pull request only"
|
||||
)
|
||||
arg_parser.add_argument(
|
||||
"--check-commits",
|
||||
action="store_true",
|
||||
help="Check and compare git commit email with GitHub account email",
|
||||
)
|
||||
args, unknown_args = arg_parser.parse_known_args()
|
||||
|
||||
Config(args.cfg_file, unknown_args)
|
||||
gh_api = github_api.GithubOrgApi()
|
||||
|
||||
if args.pr:
|
||||
pulls = [gh_api.repo.get_pull(int(args.pr))]
|
||||
else:
|
||||
pulls = gh_api.repo.get_pulls(state=args.pr_state)
|
||||
print(f"\nPRs count ({args.pr_state}):", pulls.totalCount)
|
||||
|
||||
if args.newer:
|
||||
pr_created_after = (
|
||||
datetime.datetime.now() - datetime.timedelta(minutes=int(args.newer))
|
||||
).astimezone()
|
||||
print("Checking PRs created after:", pr_created_after)
|
||||
|
||||
non_org_intel_pr_users = set()
|
||||
non_org_pr_users = set()
|
||||
wrong_pulls = {}
|
||||
|
||||
for pull in pulls:
|
||||
pr_created_at = pull.created_at.replace(tzinfo=datetime.timezone.utc).astimezone()
|
||||
if args.newer and pr_created_at <= pr_created_after:
|
||||
print(f"\nIGNORE: {get_pr_info_str(pull)}")
|
||||
continue
|
||||
|
||||
print(f"\n{get_pr_info_str(pull)}")
|
||||
if args.check_commits:
|
||||
wrong_commits = get_wrong_commits(pull)
|
||||
if wrong_commits:
|
||||
wrong_pulls[pull.number] = wrong_commits
|
||||
else:
|
||||
update_labels(gh_api, pull, non_org_intel_pr_users, non_org_pr_users)
|
||||
|
||||
if wrong_pulls:
|
||||
for pull_number, wrong_commits in wrong_pulls.items():
|
||||
print(
|
||||
f"\nERROR: Remove or replace wrong commits in the PR {pull_number}:\n ",
|
||||
"\n ".join(wrong_commits),
|
||||
)
|
||||
print(
|
||||
"\nAbout commit signature verification:\n ",
|
||||
"https://docs.github.com/en/github/authenticating-to-github/"
|
||||
"managing-commit-signature-verification/about-commit-signature-verification",
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
if non_org_intel_pr_users:
|
||||
print("\nNon org user with Intel email or company:")
|
||||
github_api.print_users(non_org_intel_pr_users)
|
||||
if non_org_pr_users:
|
||||
print("\nNon org user with NO Intel email or company:")
|
||||
github_api.print_users(non_org_pr_users)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,45 @@
|
||||
{
|
||||
"GITHUB_TOKEN": "<Put token here or set as arg or as env variable>",
|
||||
"GITHUB_ORGANIZATION": "openvinotoolkit",
|
||||
"GITHUB_REPO": "openvino",
|
||||
"IGNORE_LOGINS": [
|
||||
"openvino-ci",
|
||||
"openvino-pushbot",
|
||||
"lab-nerval",
|
||||
"lab-nerval-onnx-ci",
|
||||
"onnx-watchdog-agent",
|
||||
"workbench-ci-bot",
|
||||
"openvino-pot-ci"
|
||||
],
|
||||
"MAX_MEMBERS_TO_REMOVE": 15,
|
||||
"EMAILS_FILE_PATH": "dev_emails-test.txt",
|
||||
"PROXIES": {
|
||||
"HTTP_PROXY": null,
|
||||
"HTTPS_PROXY": null,
|
||||
"NO_PROXY": "localhost,127.0.0.1,.intel.com"
|
||||
},
|
||||
"DRY_RUN": false,
|
||||
"TEAM_TO_LABEL": {
|
||||
"openvino-admins": "category: CI",
|
||||
"openvino-maintainers": "category: IE common",
|
||||
"openvino-docs-maintainers": "category: docs",
|
||||
"openvino-ie-maintainers": "category: IE common",
|
||||
"openvino-ie-cpu-maintainers": "category: CPU",
|
||||
"openvino-ie-gna-maintainers": "category: GNA",
|
||||
"openvino-ie-gpu-maintainers": "category: GPU",
|
||||
"openvino-ie-lpt-maintainers": "category: LP transformations",
|
||||
"openvino-ie-multi-maintainers": "category: MULTI",
|
||||
"openvino-ie-python-api-maintainers": "category: python api",
|
||||
"openvino-ie-template-maintainers": "category: TEMPLATE",
|
||||
"openvino-ie-tests-maintainers": "category: IE Tests",
|
||||
"openvino-ie-vpu-maintainers": "category: VPU",
|
||||
"openvino-mo-maintainers": "category: MO",
|
||||
"openvino-ngraph-maintainers": "category: nGraph",
|
||||
"openvino-scripts-maintainers": "category: build",
|
||||
"openvino-tests-maintainers": "category: IE Tests",
|
||||
"openvino-tools-maintainers": "category: tools",
|
||||
"openvino-pot-maintainers": "category: POT",
|
||||
"openvino-configuration-mgmt": "category: dependency_changes",
|
||||
"openvino-samples-maintainers": "category: samples"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
# Copyright (C) 2018-2021 Intel Corporation
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
"""
|
||||
Configurations management
|
||||
"""
|
||||
|
||||
# pylint: disable=fixme,broad-except
|
||||
|
||||
import os
|
||||
import sys
|
||||
import ast
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
if sys.hexversion < 0x3060000:
|
||||
raise Exception("Python version must be >= 3.6")
|
||||
|
||||
|
||||
class ConfigException(Exception):
|
||||
"""Base configuration exception"""
|
||||
|
||||
|
||||
class Config:
|
||||
"""Configuration wrapper"""
|
||||
|
||||
_instance = None
|
||||
_properties = None
|
||||
|
||||
default_cfg_path = Path(__file__).resolve().parent / "config.json"
|
||||
|
||||
def __new__(cls, *_args, **_kwargs):
|
||||
if not Config._instance:
|
||||
Config._instance = super(Config, cls).__new__(cls)
|
||||
return Config._instance
|
||||
|
||||
def __init__(self, file_path=None, cli_args=None):
|
||||
"""
|
||||
:param file_path: Path to json configuration file
|
||||
:type file_path: String
|
||||
|
||||
:param args: List of argparse arguments with patterns: 'name=value' or 'name'
|
||||
:type args: list
|
||||
"""
|
||||
if Config._properties:
|
||||
return
|
||||
|
||||
self._file_path = file_path or Config.default_cfg_path
|
||||
self._cli_args = cli_args or []
|
||||
|
||||
self._json_cfg = {}
|
||||
self._args = {}
|
||||
|
||||
self._load_cfg()
|
||||
self._parse_cli_args()
|
||||
|
||||
Config._properties = {}
|
||||
for name, value in self._json_cfg.items():
|
||||
if hasattr(self, name):
|
||||
raise ConfigException(f"Duplicating prosperity: {name}")
|
||||
property_value = self._args.get(name) or os.getenv(name)
|
||||
if property_value:
|
||||
# Try to set prosperity_value as Python literal structures, e.g. DRY_RUN=False
|
||||
try:
|
||||
property_value = ast.literal_eval(property_value)
|
||||
except Exception:
|
||||
pass
|
||||
if not isinstance(property_value, type(value)):
|
||||
raise ConfigException(f"Python type of {name} parameter must be {type(value)}")
|
||||
else:
|
||||
property_value = value
|
||||
Config._properties[name] = property_value
|
||||
|
||||
self.set_proxy()
|
||||
|
||||
def __getattr__(self, attr_name):
|
||||
if attr_name in self._properties:
|
||||
return self._properties.get(attr_name)
|
||||
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{attr_name}'")
|
||||
|
||||
def _load_cfg(self):
|
||||
"""Load the json configuration file"""
|
||||
try:
|
||||
with open(self._file_path, encoding="utf-8") as conf:
|
||||
self._json_cfg = json.load(conf)
|
||||
except Exception as exc:
|
||||
raise ConfigException("Failed to load configuration from:", self._file_path) from exc
|
||||
|
||||
def _parse_cli_args(self):
|
||||
"""Parse argparse arguments with patterns: 'name=value' or 'name'"""
|
||||
for cli_arg in self._cli_args:
|
||||
arg = cli_arg.split("=")
|
||||
if arg[0] not in self._json_cfg:
|
||||
raise ConfigException(f"Unsupported argument: {arg}")
|
||||
self._args[arg[0]] = True if len(arg) == 1 else "=".join(arg[1:])
|
||||
|
||||
@property
|
||||
def properties(self):
|
||||
"""Get all properties as Dict"""
|
||||
return self._properties
|
||||
|
||||
def set_proxy(self):
|
||||
"""Set proxies"""
|
||||
for proxy_name, url in self._properties["PROXIES"].items():
|
||||
if url is not None:
|
||||
print(f"Set proxy: {proxy_name}={url}")
|
||||
os.environ[proxy_name] = url
|
||||
|
||||
|
||||
def _test():
|
||||
"""Test and debug"""
|
||||
print("Config.default_cfg_path:", Config.default_cfg_path)
|
||||
cfg = Config(cli_args=["DRY_RUN", 'PROXIES={"NO_PROXY": "localhost"}'])
|
||||
print("Config.properties:", cfg.properties)
|
||||
print("cfg.PROXIES:", cfg.PROXIES)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
_test()
|
||||
@@ -0,0 +1,9 @@
|
||||
# good comment
|
||||
Last_name, First_name <first_name.last_name@intel.com>
|
||||
first_name.last_name@intel.com
|
||||
openvino_pushbot@intel.com
|
||||
|
||||
# Wrong emails
|
||||
foo@foo.com
|
||||
foo1 foo2
|
||||
foo1 foo2@intel.com
|
||||
@@ -0,0 +1,384 @@
|
||||
# Copyright (C) 2018-2021 Intel Corporation
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
"""
|
||||
GitHub API for controlling organization
|
||||
"""
|
||||
|
||||
# pylint: disable=fixme,no-member
|
||||
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import typing
|
||||
from pathlib import Path
|
||||
|
||||
from github import Github, GithubException, RateLimitExceededException, IncompletableObject
|
||||
from github.PaginatedList import PaginatedList
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parents[1]))
|
||||
from github_org_control.configs import Config
|
||||
|
||||
|
||||
class GithubApiException(Exception):
|
||||
"""Base GitHub API exception"""
|
||||
|
||||
|
||||
def is_valid_user(user):
|
||||
"""Checks that user is valid github.Github object"""
|
||||
try:
|
||||
return user and user.login
|
||||
except IncompletableObject:
|
||||
return False
|
||||
|
||||
|
||||
def is_user_ignored(user):
|
||||
"""Checks that user should be ignored"""
|
||||
if is_valid_user(user) and user.login.lower() not in Config().IGNORE_LOGINS:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def is_valid_name(name):
|
||||
"""Checks that GitHub user's name is valid"""
|
||||
return name and len(name) >= 3 and " " in name
|
||||
|
||||
|
||||
def is_intel_email(email):
|
||||
"""Checks that email is valid Intel email"""
|
||||
return email and len(email) > 10 and " " not in email and email.lower().endswith("@intel.com")
|
||||
|
||||
|
||||
def is_intel_company(company):
|
||||
"""Checks that company contains intel"""
|
||||
return company and "intel" in company.lower()
|
||||
|
||||
|
||||
def is_valid_intel_user(user):
|
||||
"""Checks that user is valid GitHub and Intel user"""
|
||||
try:
|
||||
return is_valid_user(user) and is_valid_name(user.name) and is_intel_email(user.email)
|
||||
except IncompletableObject:
|
||||
return False
|
||||
|
||||
|
||||
def print_users(users):
|
||||
"""Print list of users in different formats: list, set, PaginatedList"""
|
||||
if isinstance(users, (list, set, PaginatedList)):
|
||||
users_count = users.totalCount if isinstance(users, PaginatedList) else len(users)
|
||||
print(f"GitHub users {users_count} (login - name - company - email - valid):")
|
||||
else:
|
||||
users = [users]
|
||||
for user in users:
|
||||
if not is_valid_user(user):
|
||||
print("WRONG GitHub user: ???")
|
||||
continue
|
||||
|
||||
try:
|
||||
name = user.name
|
||||
except IncompletableObject:
|
||||
name = "???"
|
||||
|
||||
try:
|
||||
company = user.company
|
||||
except IncompletableObject:
|
||||
company = "???"
|
||||
|
||||
try:
|
||||
email = user.email
|
||||
except IncompletableObject:
|
||||
email = "???"
|
||||
|
||||
valid_check = "OK" if is_valid_intel_user(user) else "FIX"
|
||||
if not is_intel_email(email):
|
||||
valid_check += " email"
|
||||
if not is_valid_name(name):
|
||||
valid_check += " name"
|
||||
print(f'{user.login} - "{name}" - "{company}" - {email} - {valid_check}')
|
||||
|
||||
|
||||
def get_dev_emails():
|
||||
"""
|
||||
Read a file with developer emails. Supported email formats
|
||||
first_name.last_name@intel.com
|
||||
Import from Outlook: Last_name, First_name <first_name.last_name@intel.com>
|
||||
"""
|
||||
re_email = re.compile(r".+<(.+)>")
|
||||
emails = set()
|
||||
cfg = Config()
|
||||
with open(cfg.properties["EMAILS_FILE_PATH"]) as file_obj:
|
||||
for line in file_obj:
|
||||
line = line.strip().lower()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
re_outlook_email = re_email.match(line)
|
||||
if re_outlook_email:
|
||||
line = re_outlook_email.group(1).strip()
|
||||
if not is_intel_email(line):
|
||||
print(f'Wrong email in {cfg.properties["EMAILS_FILE_PATH"]}: {line}')
|
||||
continue
|
||||
emails.add(line)
|
||||
return emails
|
||||
|
||||
|
||||
class GithubOrgApi:
|
||||
"""Common API for GitHub organization"""
|
||||
|
||||
def __init__(self):
|
||||
self._cfg = Config()
|
||||
self.github = Github(self._cfg.GITHUB_TOKEN)
|
||||
self.github_org = self.github.get_organization(self._cfg.GITHUB_ORGANIZATION)
|
||||
self.repo = self.github.get_repo(f"{self._cfg.GITHUB_ORGANIZATION}/{self._cfg.GITHUB_REPO}")
|
||||
self.github_users_by_email = {}
|
||||
self.org_members_by_login = {}
|
||||
self.members_to_remove = set()
|
||||
self.members_to_fix_name = set()
|
||||
|
||||
def is_org_user(self, user):
|
||||
"""Checks that user is a member of GitHub organization"""
|
||||
if is_valid_user(user):
|
||||
# user.get_organization_membership(self.github_org) doesn't work with org members
|
||||
# permissions, GITHUB_TOKEN must be org owner now
|
||||
return self.github_org.has_in_members(user)
|
||||
return False
|
||||
|
||||
def get_org_emails(self):
|
||||
"""Gets and prints emails of all GitHub organization members"""
|
||||
org_members = self.github_org.get_members()
|
||||
org_emails = set()
|
||||
|
||||
print(f"\nOrg members {org_members.totalCount} (login - name - company - email - valid):")
|
||||
for org_member in org_members:
|
||||
self.org_members_by_login[org_member.login.lower()] = org_member
|
||||
print_users(org_member)
|
||||
if is_intel_email(org_member.email):
|
||||
email = org_member.email.lower()
|
||||
org_emails.add(email)
|
||||
self.github_users_by_email[email] = org_member
|
||||
if not is_valid_name(org_member.name):
|
||||
self.members_to_fix_name.add(org_member)
|
||||
elif not is_user_ignored(org_member):
|
||||
self.members_to_remove.add(org_member)
|
||||
|
||||
print("\nOrg members - no Intel emails:")
|
||||
print_users(self.members_to_remove)
|
||||
|
||||
print("\nOrg members - no real name:")
|
||||
print_users(self.members_to_fix_name)
|
||||
print(
|
||||
"\nOrg member emails - no real name:",
|
||||
"; ".join([member.email.lower() for member in self.members_to_fix_name]),
|
||||
)
|
||||
|
||||
return org_emails
|
||||
|
||||
def get_org_invitation_emails(self):
|
||||
"""Gets GitHub organization teams prints info"""
|
||||
org_invitations = self.github_org.invitations()
|
||||
org_invitation_emails = set()
|
||||
|
||||
print(
|
||||
f"\nOrg invitations {org_invitations.totalCount} "
|
||||
"(login - name - company - email - valid):"
|
||||
)
|
||||
for org_invitation in org_invitations:
|
||||
print_users(org_invitation)
|
||||
if is_user_ignored(org_invitation):
|
||||
continue
|
||||
if is_intel_email(org_invitation.email):
|
||||
org_invitation_emails.add(org_invitation.email.lower())
|
||||
else:
|
||||
print("Strange org invitation:", org_invitation)
|
||||
|
||||
print(
|
||||
f"\nOrg invitation emails {len(org_invitation_emails)}:",
|
||||
"; ".join(org_invitation_emails),
|
||||
)
|
||||
return org_invitation_emails
|
||||
|
||||
def get_org_teams(self):
|
||||
"""Gets GitHub organization teams prints info"""
|
||||
teams = []
|
||||
org_teams = self.github_org.get_teams()
|
||||
print("\nOrg teams count:", org_teams.totalCount)
|
||||
for team in org_teams:
|
||||
teams.append(team.name)
|
||||
print(f"\nTeam: {team.name} - parent: {team.parent}")
|
||||
|
||||
repos = team.get_repos()
|
||||
print("Repos:")
|
||||
for repo in repos:
|
||||
print(f" {repo.name} -", team.get_repo_permission(repo))
|
||||
|
||||
team_maintainers = team.get_members(role="maintainer")
|
||||
team_maintainer_logins = set()
|
||||
for maintainer in team_maintainers:
|
||||
team_maintainer_logins.add(maintainer.login)
|
||||
team_members = team.get_members(role="member")
|
||||
team_member_logins = set()
|
||||
for member in team_members:
|
||||
team_member_logins.add(member.login)
|
||||
members = team.get_members(role="all")
|
||||
member_emails = []
|
||||
print("Members (role - login - name - company - email - valid):")
|
||||
for user in members:
|
||||
if user.login in team_maintainer_logins:
|
||||
print(" Maintainer - ", end="")
|
||||
elif user.login in team_member_logins:
|
||||
print(" Member - ", end="")
|
||||
else:
|
||||
# It is not possible to check child teams members
|
||||
print(" ??? - ", end="")
|
||||
print_users(user)
|
||||
if is_intel_email(user.email) and not is_user_ignored(user):
|
||||
member_emails.append(user.email.lower())
|
||||
print(f"Intel emails {len(member_emails)}:", "; ".join(member_emails))
|
||||
return teams
|
||||
|
||||
def get_github_user_by_email(self, email):
|
||||
"""Gets GitHub user by email"""
|
||||
if email in self.github_users_by_email:
|
||||
return self.github_users_by_email.get(email)
|
||||
|
||||
def search_users():
|
||||
paginated_users = self.github.search_users(f"{email} in:email")
|
||||
# Minimize the GitHub Rate Limit
|
||||
users = []
|
||||
for user in paginated_users:
|
||||
users.append(user)
|
||||
if len(users) == 1:
|
||||
return users[0]
|
||||
if len(users) == 0:
|
||||
return None
|
||||
raise GithubApiException(
|
||||
f"ERROR: Found {len(users)} GitHub accounts with the same email {email}"
|
||||
)
|
||||
|
||||
try:
|
||||
user = search_users()
|
||||
except RateLimitExceededException:
|
||||
print("WARNING: RateLimitExceededException")
|
||||
time.sleep(30)
|
||||
user = search_users()
|
||||
self.github_users_by_email[email] = user
|
||||
|
||||
return user
|
||||
|
||||
def get_valid_github_users(self, emails):
|
||||
"""Gets valid GitHub users by email and prints status"""
|
||||
valid_users = set()
|
||||
wrong_emails = set()
|
||||
no_account_emails = set()
|
||||
no_account_names = set()
|
||||
print(f"\nGitHub users from {len(emails)} invite emails (email - status):")
|
||||
for email in emails:
|
||||
if not is_intel_email(email):
|
||||
print(f"{email} - Non Intel email")
|
||||
wrong_emails.add(email)
|
||||
continue
|
||||
|
||||
# You can make up to 30 requests per minute; https://developer.github.com/v3/search/
|
||||
time.sleep(2)
|
||||
user = self.get_github_user_by_email(email)
|
||||
|
||||
if not user:
|
||||
print(f"{email} - No valid GitHub account")
|
||||
no_account_emails.add(email)
|
||||
continue
|
||||
|
||||
if user.email and user.email.lower() == email:
|
||||
if is_valid_name(user.name):
|
||||
print(f"{email} - OK")
|
||||
valid_users.add(user)
|
||||
else:
|
||||
print(f"{email} - No valid name in GitHub account: ", end="")
|
||||
print_users(user)
|
||||
no_account_names.add(email)
|
||||
else:
|
||||
print(f"{email} - Non public or wrong email in GitHub account: ", end="")
|
||||
print_users(user)
|
||||
no_account_emails.add(email)
|
||||
|
||||
print("\nValid users:")
|
||||
print_users(valid_users)
|
||||
|
||||
print(f"\nWrong emails {len(wrong_emails)}:", "; ".join(wrong_emails))
|
||||
|
||||
print(
|
||||
f"\nIntel emails - No valid GitHub account {len(no_account_emails)}:",
|
||||
"; ".join(no_account_emails),
|
||||
)
|
||||
|
||||
print(
|
||||
f"\nIntel emails - No valid name in GitHub account {len(no_account_names)}:",
|
||||
"; ".join(no_account_names),
|
||||
)
|
||||
return valid_users
|
||||
|
||||
def invite_users(self, users):
|
||||
"""Invites users to GitHub organization and prints status"""
|
||||
if not isinstance(users, typing.Iterable):
|
||||
users = [users]
|
||||
print(f"\nInvite {len(users)} users:")
|
||||
|
||||
for user in users:
|
||||
if isinstance(user, str):
|
||||
print(f"Email: {user}")
|
||||
self.github_org.invite_user(email=user)
|
||||
else:
|
||||
print(f'{user.login} - "{user.name}" - {user.email} - ', end="")
|
||||
try:
|
||||
if is_user_ignored(user):
|
||||
print("Ignored")
|
||||
continue
|
||||
if self._cfg.DRY_RUN:
|
||||
print("Dry run")
|
||||
continue
|
||||
self.github_org.invite_user(user=user)
|
||||
print("OK")
|
||||
except GithubException as exc:
|
||||
print(f'FAIL: {exc.data["errors"][0]["message"]}')
|
||||
|
||||
def remove_users(self, users):
|
||||
"""Removes users from GitHub organization"""
|
||||
if not isinstance(users, typing.Iterable):
|
||||
users = [users]
|
||||
print(f"\nRemove {len(users)} users:")
|
||||
|
||||
dry_run = self._cfg.DRY_RUN
|
||||
if not dry_run and len(users) > self._cfg.MAX_MEMBERS_TO_REMOVE:
|
||||
print(
|
||||
"WARNING: Review is required for removing members more than "
|
||||
f"{self._cfg.MAX_MEMBERS_TO_REMOVE}"
|
||||
)
|
||||
# TODO: Add notification
|
||||
dry_run = True
|
||||
|
||||
for user in users:
|
||||
member = self.get_github_user_by_email(user) if isinstance(user, str) else user
|
||||
print(f'{member.login} - "{member.name}" - {member.email} - ', end="")
|
||||
try:
|
||||
if is_user_ignored(member):
|
||||
print("Ignored")
|
||||
continue
|
||||
if dry_run:
|
||||
print("Dry run")
|
||||
continue
|
||||
self.github_org.remove_from_membership(member)
|
||||
print("OK")
|
||||
except GithubException as exc:
|
||||
print(f'FAIL: {exc.data["errors"][0]["message"]}')
|
||||
|
||||
|
||||
def _test():
|
||||
"""Test and debug"""
|
||||
Config(cli_args=["DRY_RUN=True"])
|
||||
dev_emails = get_dev_emails()
|
||||
print("dev_emails:", dev_emails)
|
||||
|
||||
gh_api = GithubOrgApi()
|
||||
gh_api.get_org_emails()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
_test()
|
||||
@@ -0,0 +1,247 @@
|
||||
# Copyright (C) 2018-2021 Intel Corporation
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
"""
|
||||
Gets info about users and groups via LDAP
|
||||
"""
|
||||
|
||||
# pylint: disable=fixme,no-member
|
||||
|
||||
import sys
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
|
||||
from ldap3 import Server, Connection, ALL, SUBTREE
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parents[1]))
|
||||
from github_org_control.configs import Config
|
||||
|
||||
|
||||
class LdapApiException(Exception):
|
||||
"""Base LDAP API exception"""
|
||||
|
||||
|
||||
class InfoLevel(Enum):
|
||||
"""Constants for printing user info from LDAP"""
|
||||
|
||||
PDL = "PDL" # Public Distribution List (group of e-mail addresses)
|
||||
FULL = "Full"
|
||||
|
||||
|
||||
def print_user_info(info, info_level=None):
|
||||
"""Pretty-print of a user info data structure (dict). info_level is the InfoLevel Enum"""
|
||||
if not info or not info.get("mail"):
|
||||
raise LdapApiException("ERROR: No info or absent mail")
|
||||
|
||||
def get_membership():
|
||||
if info_level == InfoLevel.PDL:
|
||||
membership_info = " PDLs:"
|
||||
elif info_level == InfoLevel.FULL:
|
||||
membership_info = " memberOf :"
|
||||
else:
|
||||
return ""
|
||||
# Grouping groups by purpose
|
||||
if info_level == InfoLevel.PDL:
|
||||
sort_key = lambda i: i.split(",", 1)[0].lower()
|
||||
else:
|
||||
sort_key = lambda i: i.split(",", 1)[1] + i.split(",", 1)[0].lower()
|
||||
for item in sorted(info["memberOf"], key=sort_key):
|
||||
if info_level == InfoLevel.PDL and "OU=Delegated" not in item:
|
||||
continue
|
||||
membership_info += f"\n {item}"
|
||||
return membership_info
|
||||
|
||||
try:
|
||||
text_info = (
|
||||
f'\n{info["cn"]} <{info["mail"]}>; {info["sAMAccountName"]}; {info["employeeID"]}'
|
||||
f'\n Org group: {info["intelSuperGroupDescr"]} ({info["intelSuperGroupShortName"]}) /'
|
||||
f' {info["intelGroupDescr"]} ({info["intelGroupShortName"]}) /'
|
||||
f' {info["intelDivisionDescr"]} ({info["intelDivisionShortName"]}) /'
|
||||
f' {info["intelOrgUnitDescr"]}'
|
||||
f'\n Manager: {info.get("manager")}'
|
||||
f'\n Location: {info["intelRegionCode"]} / {info["co"]} / {info["intelSiteCode"]} /'
|
||||
f' {info["intelBldgCode"]} ({info.get("intelSiteName")}) /'
|
||||
f' {info["physicalDeliveryOfficeName"]}'
|
||||
f'\n Other: {info["employeeType"]} | {info["intelExportCountryGroup"]} |'
|
||||
f' {info["whenCreated"]} | {info["intelCostCenterDescr"]} | {info["jobDescription"]}'
|
||||
)
|
||||
except Exception as exc:
|
||||
raise LdapApiException(
|
||||
f'ERROR: Failed to get info about "{info["mail"]}". '
|
||||
f"Exception occurred:\n{repr(exc)}"
|
||||
) from exc
|
||||
print(text_info)
|
||||
|
||||
membership = get_membership()
|
||||
if info_level == InfoLevel.PDL and membership:
|
||||
print(membership)
|
||||
elif info_level == InfoLevel.FULL:
|
||||
for key in sorted(info):
|
||||
if isinstance(info[key], list):
|
||||
if key == "memberOf":
|
||||
print(membership)
|
||||
else:
|
||||
print(f" {key} :")
|
||||
for item in info[key]:
|
||||
print(" ", item)
|
||||
else:
|
||||
print(f" {key} : {info[key]}")
|
||||
|
||||
|
||||
class LdapApi:
|
||||
"""LDAP API for getting user info and emails"""
|
||||
|
||||
_binary_blobs = ["thumbnailPhoto", "msExchUMSpokenName", "msExchBlockedSendersHash"]
|
||||
_check_existing = [
|
||||
"intelExportCountryGroup",
|
||||
"physicalDeliveryOfficeName",
|
||||
"intelSuperGroupShortName",
|
||||
"intelGroupShortName",
|
||||
"intelDivisionShortName",
|
||||
]
|
||||
|
||||
null = "<null>"
|
||||
|
||||
def __init__(self):
|
||||
self._cfg = Config()
|
||||
self.server = Server(self._cfg.LDAP_SERVER, get_info=ALL)
|
||||
self.connection = Connection(
|
||||
self.server, user=self._cfg.LDAP_USER, password=self._cfg.LDAP_PASSWORD, auto_bind=True
|
||||
)
|
||||
self.connection.bind()
|
||||
|
||||
def get_user_emails(self, groups=None):
|
||||
"""Gets emails of LDAP groups and sub-groups"""
|
||||
print("\nGet emails from LDAP groups:")
|
||||
processed_ldap_members = {}
|
||||
|
||||
def process_group_members(member, parent_group):
|
||||
if member in processed_ldap_members:
|
||||
processed_ldap_members[member]["parent_groups"].append(parent_group)
|
||||
print(
|
||||
"\nWARNING: Ignore LDAP member to avoid duplication and recursive cycling "
|
||||
f"of PDLs: {member}\n "
|
||||
f'email: {processed_ldap_members[member].get("email")}\n parent_groups:'
|
||||
)
|
||||
for group in processed_ldap_members[member].get("parent_groups", []):
|
||||
print(7 * " ", group)
|
||||
|
||||
return
|
||||
processed_ldap_members[member] = {"email": None, "parent_groups": [parent_group]}
|
||||
|
||||
# AD moves terminated users to the boneyard OU in case the user returns,
|
||||
# so it can be reactivated with little effort.
|
||||
# After 30 days it is removed and the unix personality becomes unlinked.
|
||||
if "OU=Boneyard" in member:
|
||||
return
|
||||
self.connection.search(
|
||||
member, r"(objectClass=*)", SUBTREE, attributes=["cn", "member", "mail"]
|
||||
)
|
||||
|
||||
# print(self.connection.entries)
|
||||
if not self.connection.response:
|
||||
raise LdapApiException(f"ERROR: empty response. LDAP member: {member}")
|
||||
|
||||
# Check that the member is worker.
|
||||
# The response can contain several items, but the first item is valid only
|
||||
if "OU=Workers" in member:
|
||||
if self.connection.response[0]["attributes"]["mail"]:
|
||||
processed_ldap_members[member]["email"] = self.connection.response[0][
|
||||
"attributes"
|
||||
]["mail"].lower()
|
||||
return
|
||||
raise LdapApiException(
|
||||
f"ERROR: no mail. LDAP worker: {member}\n" f"{self.connection.entries}"
|
||||
)
|
||||
|
||||
if len(self.connection.response) > 1:
|
||||
raise LdapApiException(
|
||||
f"ERROR: multiple responses for {member}: "
|
||||
f"{len(self.connection.response)}\n"
|
||||
f"{self.connection.entries}"
|
||||
)
|
||||
|
||||
if self.connection.response[0]["attributes"]["member"]:
|
||||
for group_member in self.connection.response[0]["attributes"]["member"]:
|
||||
process_group_members(group_member, member)
|
||||
else:
|
||||
print(f"\nERROR: no members in LDAP group: {member}\n{self.connection.entries}")
|
||||
|
||||
for group in groups or self._cfg.LDAP_PDLs:
|
||||
print("\nProcess ROOT LDAP group:", group)
|
||||
process_group_members(group, "ROOT")
|
||||
return {
|
||||
member.get("email") for member in processed_ldap_members.values() if member.get("email")
|
||||
}
|
||||
|
||||
def _get_user_info(self, query):
|
||||
"""Gets user info from LDAP as dict matching key and values pairs from query"""
|
||||
query_filter = "".join(f"({key}={value})" for key, value in query.items())
|
||||
|
||||
for domain in self._cfg.LDAP_DOMAINS:
|
||||
search_base = f"OU=Workers,DC={domain},DC=corp,DC=intel,DC=com"
|
||||
self.connection.search(
|
||||
search_base,
|
||||
f"(&(objectcategory=person)(objectclass=user)(intelflags=1){query_filter})",
|
||||
SUBTREE,
|
||||
attributes=["*"],
|
||||
)
|
||||
|
||||
if self.connection.response:
|
||||
if len(self.connection.response) > 1:
|
||||
raise LdapApiException(
|
||||
f"ERROR: multiple responses for {query_filter}: "
|
||||
f"{len(self.connection.response)}\n"
|
||||
f"{self.connection.entries}"
|
||||
)
|
||||
info = self.connection.response[0]["attributes"]
|
||||
|
||||
# remove long binary blobs
|
||||
for blob in LdapApi._binary_blobs:
|
||||
info[blob] = b""
|
||||
for key in LdapApi._check_existing:
|
||||
if not info.get(key):
|
||||
info[key] = LdapApi.null
|
||||
return info
|
||||
return {}
|
||||
|
||||
def get_user_info_by_idsid(self, idsid):
|
||||
"""Gets user info from LDAP as dict using account name for searching"""
|
||||
return self._get_user_info({"sAMAccountName": idsid})
|
||||
|
||||
def get_user_info_by_name(self, name):
|
||||
"""Gets user info from LDAP as dict using common name for searching"""
|
||||
return self._get_user_info({"cn": name})
|
||||
|
||||
def get_user_info_by_email(self, email):
|
||||
"""Gets user info from LDAP as dict using emails for searching"""
|
||||
return self._get_user_info({"mail": email})
|
||||
|
||||
def get_absent_emails(self, emails):
|
||||
"""Checks users by email in LDAP and returns absent emails"""
|
||||
absent_emails = set()
|
||||
for email in emails:
|
||||
if not self.get_user_info_by_email(email):
|
||||
absent_emails.add(email)
|
||||
return absent_emails
|
||||
|
||||
|
||||
def _test():
|
||||
"""Test and debug"""
|
||||
ldap = LdapApi()
|
||||
|
||||
emails = ldap.get_user_emails()
|
||||
print(f'\nLDAP emails count: {len(emails)}\n{"; ".join(emails)}')
|
||||
|
||||
emails = ["foo@intel.com"]
|
||||
|
||||
for email in emails:
|
||||
info = ldap.get_user_info_by_email(email)
|
||||
if info:
|
||||
print_user_info(info, InfoLevel.PDL)
|
||||
else:
|
||||
print(f"\n{email} - not found")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
_test()
|
||||
@@ -0,0 +1 @@
|
||||
pylint==2.11.1
|
||||
@@ -0,0 +1,2 @@
|
||||
PyGithub==1.55
|
||||
ldap3==2.7
|
||||
Reference in new issue
Block a user