mirror of
https://github.com/netbox-community/netbox.git
synced 2026-01-22 11:38:45 -06:00
Compare commits
2 Commits
20902-git-
...
21249-omit
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4c340b6353 | ||
|
|
062b54c6c0 |
@@ -102,10 +102,7 @@ class GitBackend(DataBackend):
|
|||||||
clone_args['pool_manager'] = ProxyPoolManager(self.socks_proxy)
|
clone_args['pool_manager'] = ProxyPoolManager(self.socks_proxy)
|
||||||
|
|
||||||
if self.url_scheme in ('http', 'https'):
|
if self.url_scheme in ('http', 'https'):
|
||||||
# Only pass explicit credentials if URL doesn't already contain embedded username
|
if self.params.get('username'):
|
||||||
# to avoid credential conflicts
|
|
||||||
parsed_url = urlparse(self.url)
|
|
||||||
if not parsed_url.username and self.params.get('username'):
|
|
||||||
clone_args.update(
|
clone_args.update(
|
||||||
{
|
{
|
||||||
"username": self.params.get('username'),
|
"username": self.params.get('username'),
|
||||||
|
|||||||
@@ -1,59 +0,0 @@
|
|||||||
from unittest.mock import patch
|
|
||||||
|
|
||||||
from django.test import TestCase
|
|
||||||
|
|
||||||
from core.data_backends import GitBackend
|
|
||||||
|
|
||||||
|
|
||||||
class GitBackendCredentialTests(TestCase):
|
|
||||||
|
|
||||||
def _get_clone_kwargs(self, url, **params):
|
|
||||||
backend = GitBackend(url=url, **params)
|
|
||||||
|
|
||||||
with patch('dulwich.porcelain.clone') as mock_clone, \
|
|
||||||
patch('dulwich.porcelain.NoneStream'):
|
|
||||||
try:
|
|
||||||
with backend.fetch():
|
|
||||||
pass
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
if mock_clone.called:
|
|
||||||
return mock_clone.call_args.kwargs
|
|
||||||
return {}
|
|
||||||
|
|
||||||
def test_url_with_embedded_username_skips_explicit_credentials(self):
|
|
||||||
kwargs = self._get_clone_kwargs(
|
|
||||||
url='https://myuser@bitbucket.org/workspace/repo.git',
|
|
||||||
username='myuser',
|
|
||||||
password='my-api-key'
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertEqual(kwargs.get('username'), None)
|
|
||||||
self.assertEqual(kwargs.get('password'), None)
|
|
||||||
|
|
||||||
def test_url_without_embedded_username_passes_explicit_credentials(self):
|
|
||||||
kwargs = self._get_clone_kwargs(
|
|
||||||
url='https://bitbucket.org/workspace/repo.git',
|
|
||||||
username='myuser',
|
|
||||||
password='my-api-key'
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertEqual(kwargs.get('username'), 'myuser')
|
|
||||||
self.assertEqual(kwargs.get('password'), 'my-api-key')
|
|
||||||
|
|
||||||
def test_url_with_embedded_username_no_explicit_credentials(self):
|
|
||||||
kwargs = self._get_clone_kwargs(
|
|
||||||
url='https://myuser@bitbucket.org/workspace/repo.git'
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertEqual(kwargs.get('username'), None)
|
|
||||||
self.assertEqual(kwargs.get('password'), None)
|
|
||||||
|
|
||||||
def test_public_repo_no_credentials(self):
|
|
||||||
kwargs = self._get_clone_kwargs(
|
|
||||||
url='https://github.com/public/repo.git'
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertEqual(kwargs.get('username'), None)
|
|
||||||
self.assertEqual(kwargs.get('password'), None)
|
|
||||||
@@ -86,7 +86,7 @@ def enqueue_event(queue, instance, request, event_type):
|
|||||||
|
|
||||||
|
|
||||||
def process_event_rules(event_rules, object_type, event_type, data, username=None, snapshots=None, request=None):
|
def process_event_rules(event_rules, object_type, event_type, data, username=None, snapshots=None, request=None):
|
||||||
user = User.objects.get(username=username) if username else None
|
user = None # To be resolved from the username if needed
|
||||||
|
|
||||||
for event_rule in event_rules:
|
for event_rule in event_rules:
|
||||||
|
|
||||||
@@ -134,6 +134,10 @@ def process_event_rules(event_rules, object_type, event_type, data, username=Non
|
|||||||
# Resolve the script from action parameters
|
# Resolve the script from action parameters
|
||||||
script = event_rule.action_object.python_class()
|
script = event_rule.action_object.python_class()
|
||||||
|
|
||||||
|
# Retrieve the User if not already resolved
|
||||||
|
if user is None:
|
||||||
|
user = User.objects.get(username=username)
|
||||||
|
|
||||||
# Enqueue a Job to record the script's execution
|
# Enqueue a Job to record the script's execution
|
||||||
from extras.jobs import ScriptJob
|
from extras.jobs import ScriptJob
|
||||||
params = {
|
params = {
|
||||||
|
|||||||
Reference in New Issue
Block a user