mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

allow to pass login/password which will be used to acquire token for further usage

Browse Source
This commit is contained in:
Yaroshevich, Denis
2021-04-20 14:33:42 +03:00
parent 988e8d27fb
commit 0ff933244a
2 changed files with 137 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
octodns/provider/gcore.py
+77 -27
View File
@@ -11,7 +11,9 @@ from __future__ import (
from collections import defaultdict
from requests import Session
import http
import logging
import urllib.parse
from ..record import Record
from .base import BaseProvider
@@ -34,51 +36,82 @@ class GCoreClientNotFound(GCoreClientException):
class GCoreClient(object):
ROOT_ZONES = "/zones"
ROOT_ZONES = "zones"
def __init__(self, log, base_url, token):
session = Session()
session.headers.update({"Authorization": "Bearer {}".format(token)})
def __init__(
self,
log,
api_url,
auth_url,
token=None,
login=None,
password=None,
):
self.log = log
self._session = session
self._base_url = base_url
self._session = Session()
self._api_url = api_url
if token is not None:
self._session.headers.update(
{"Authorization": "APIKey {}".format(token)}
)
elif login is not None and password is not None:
token = self._auth(auth_url, login, password)
self._session.headers.update(
{"Authorization": "Bearer {}".format(token)}
)
else:
raise ValueError("either token or login & password must be set")
def _request(self, method, path, params={}, data=None):
url = "{}{}".format(self._base_url, path)
def _auth(self, url, login, password):
# well, can't use _request, since API returns 400 if credentials
# invalid which will be logged, but we don't want do this
r = self._session.request(
"POST",
self._build_url(url, "auth", "jwt", "login"),
json={"username": login, "password": password},
)
r.raise_for_status()
return r.json()["access"]
def _request(self, method, url, params=None, data=None):
r = self._session.request(
method, url, params=params, json=data, timeout=30.0
)
if r.status_code == 400:
if r.status_code == http.HTTPStatus.BAD_REQUEST:
self.log.error(
"bad request %r has been sent to %r: %s", data, url, r.text
)
raise GCoreClientBadRequest(r)
elif r.status_code == 404:
self.log.error(
"resource %r not found: %s", url, r.text
)
elif r.status_code == http.HTTPStatus.NOT_FOUND:
self.log.error("resource %r not found: %s", url, r.text)
raise GCoreClientNotFound(r)
elif r.status_code == 500:
self.log.error(
"server error no %r to %r: %s", data, url, r.text
)
elif r.status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR:
self.log.error("server error no %r to %r: %s", data, url, r.text)
raise GCoreClientException(r)
r.raise_for_status()
return r
def zone(self, zone_name):
return self._request(
"GET", "{}/{}".format(self.ROOT_ZONES, zone_name)
"GET", self._build_url(self._api_url, self.ROOT_ZONES, zone_name)
).json()
def zone_create(self, zone_name):
return self._request(
"POST", self.ROOT_ZONES, data={"name": zone_name}
"POST",
self._build_url(self._api_url, self.ROOT_ZONES),
data={"name": zone_name},
).json()
def zone_records(self, zone_name):
rrsets = self._request(
"GET", "{}/{}/rrsets?all=true".format(self.ROOT_ZONES, zone_name)
"GET",
"{}".format(
self._build_url(
self._api_url, self.ROOT_ZONES, zone_name, "rrsets"
)
),
params={"all": "true"},
).json()
records = rrsets["rrsets"]
return records
@@ -97,10 +130,17 @@ class GCoreClient(object):
self._request("DELETE", self._rrset_url(zone_name, rrset_name, type_))
def _rrset_url(self, zone_name, rrset_name, type_):
return "{}/{}/{}/{}".format(
self.ROOT_ZONES, zone_name, rrset_name, type_
return self._build_url(
self._api_url, self.ROOT_ZONES, zone_name, rrset_name, type_
)
@staticmethod
def _build_url(base, *items):
for i in items:
base = base.strip("/") + "/"
base = urllib.parse.urljoin(base, i)
return base
class GCoreProvider(BaseProvider):
"""
@@ -108,8 +148,12 @@ class GCoreProvider(BaseProvider):
gcore:
class: octodns.provider.gcore.GCoreProvider
# Your API key (required)
# Your API key
token: XXXXXXXXXXXX
# or login + password
login: XXXXXXXXXXXX
password: XXXXXXXXXXXX
# auth_url: https://api.gcdn.co
# url: https://dnsapi.gcorelabs.com/v2
"""
@@ -117,12 +161,18 @@ class GCoreProvider(BaseProvider):
SUPPORTS_DYNAMIC = False
SUPPORTS = set(("A", "AAAA"))
def __init__(self, id, token, *args, **kwargs):
base_url = kwargs.pop("url", "https://dnsapi.gcorelabs.com/v2")
def __init__(self, id, *args, **kwargs):
token = kwargs.pop("token", None)
login = kwargs.pop("login", None)
password = kwargs.pop("password", None)
api_url = kwargs.pop("url", "https://dnsapi.gcorelabs.com/v2")
auth_url = kwargs.pop("auth_url", "https://api.gcdn.co")
self.log = logging.getLogger("GCoreProvider[{}]".format(id))
self.log.debug("__init__: id=%s, token=***", id)
self.log.debug("__init__: id=%s", id)
super(GCoreProvider, self).__init__(id, *args, **kwargs)
self._client = GCoreClient(self.log, base_url, token)
self._client = GCoreClient(
self.log, api_url, auth_url, token, login, password
)
def _data_for_single(self, _type, record):
return {
tests/test_octodns_provider_gcore.py
+60 -12
View File
@@ -33,7 +33,7 @@ class TestGCoreProvider(TestCase):
def test_populate(self):
provider = GCoreProvider("test_id", "token")
provider = GCoreProvider("test_id", token="token")
# 400 - Bad Request.
with requests_mock() as mock:
@@ -52,7 +52,7 @@ class TestGCoreProvider(TestCase):
with self.assertRaises(GCoreClientNotFound) as ctx:
zone = Zone("unit.tests.", [])
provider._client.zone(zone)
provider._client.zone(zone.name)
self.assertIn(
'"error":"zone is not found"', text_type(ctx.exception)
)
@@ -66,6 +66,48 @@ class TestGCoreProvider(TestCase):
provider.populate(zone)
self.assertEquals("Things caught fire", text_type(ctx.exception))
# No credentials or token error
with requests_mock() as mock:
with self.assertRaises(ValueError) as ctx:
GCoreProvider("test_id")
self.assertEquals(
"either token or login & password must be set",
text_type(ctx.exception),
)
# Auth with login password
with requests_mock() as mock:
def match_body(request):
return {"username": "foo", "password": "bar"} == request.json()
auth_url = "http://api/auth/jwt/login"
mock.post(
auth_url,
additional_matcher=match_body,
status_code=200,
json={"access": "access"},
)
providerPassword = GCoreProvider(
"test_id",
url="http://dns",
auth_url="http://api",
login="foo",
password="bar",
)
assert mock.called
# make sure token passed in header
zone_rrset_url = "http://dns/zones/unit.tests/rrsets?all=true"
mock.get(
zone_rrset_url,
request_headers={"Authorization": "Bearer access"},
status_code=404,
)
zone = Zone("unit.tests.", [])
assert not providerPassword.populate(zone)
# No diffs == no changes
with requests_mock() as mock:
base = "https://dnsapi.gcorelabs.com/v2/zones/unit.tests/rrsets"
@@ -97,7 +139,7 @@ class TestGCoreProvider(TestCase):
)
def test_apply(self):
provider = GCoreProvider("test_id", "token")
provider = GCoreProvider("test_id", url="http://api", token="token")
# Zone does not exists but can be created.
with requests_mock() as mock:
@@ -153,12 +195,16 @@ class TestGCoreProvider(TestCase):
provider._client._request.assert_has_calls(
[
call("GET", "/zones/unit.tests/rrsets?all=true"),
call("GET", "/zones/unit.tests"),
call("POST", "/zones", data={"name": "unit.tests"}),
call(
"GET",
"http://api/zones/unit.tests/rrsets",
params={"all": "true"},
),
call("GET", "http://api/zones/unit.tests"),
call("POST", "http://api/zones", data={"name": "unit.tests"}),
call(
"POST",
"/zones/unit.tests/www.sub.unit.tests./A",
"http://api/zones/unit.tests/www.sub.unit.tests./A",
data={
"ttl": 300,
"resource_records": [{"content": ["2.2.3.6"]}],
@@ -166,7 +212,7 @@ class TestGCoreProvider(TestCase):
),
call(
"POST",
"/zones/unit.tests/www.unit.tests./A",
"http://api/zones/unit.tests/www.unit.tests./A",
data={
"ttl": 300,
"resource_records": [{"content": ["2.2.3.6"]}],
@@ -174,7 +220,7 @@ class TestGCoreProvider(TestCase):
),
call(
"POST",
"/zones/unit.tests/aaaa.unit.tests./AAAA",
"http://api/zones/unit.tests/aaaa.unit.tests./AAAA",
data={
"ttl": 600,
"resource_records": [
@@ -188,7 +234,7 @@ class TestGCoreProvider(TestCase):
),
call(
"POST",
"/zones/unit.tests/unit.tests./A",
"http://api/zones/unit.tests/unit.tests./A",
data={
"ttl": 300,
"resource_records": [
@@ -239,10 +285,12 @@ class TestGCoreProvider(TestCase):
provider._client._request.assert_has_calls(
[
call("DELETE", "/zones/unit.tests/www.unit.tests./A"),
call(
"DELETE", "http://api/zones/unit.tests/www.unit.tests./A"
),
call(
"PUT",
"/zones/unit.tests/ttl.unit.tests./A",
"http://api/zones/unit.tests/ttl.unit.tests./A",
data={
"ttl": 300,
"resource_records": [{"content": ["3.2.3.4"]}],