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)
|
||||
|
||||
if self.url_scheme in ('http', 'https'):
|
||||
# Only pass explicit credentials if URL doesn't already contain embedded username
|
||||
# to avoid credential conflicts
|
||||
parsed_url = urlparse(self.url)
|
||||
if not parsed_url.username and self.params.get('username'):
|
||||
if self.params.get('username'):
|
||||
clone_args.update(
|
||||
{
|
||||
"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):
|
||||
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:
|
||||
|
||||
@@ -134,6 +134,10 @@ def process_event_rules(event_rules, object_type, event_type, data, username=Non
|
||||
# Resolve the script from action parameters
|
||||
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
|
||||
from extras.jobs import ScriptJob
|
||||
params = {
|
||||
|
||||
Reference in New Issue
Block a user