Compare commits

..

1 Commits

Author SHA1 Message Date
Jason Novinger
7eedefb2df Fixes #20902: Avoid conflict when Git URL contains embedded username
When cloning Git data sources with URLs containing embedded usernames (e.g., https://user@bitbucket.org/...), NetBox was also passing explicit username/password kwargs to dulwich, causing authentication failures.

Now checks for URL-embedded credentials before passing explicit kwargs.
2026-01-21 15:24:44 -06:00
12 changed files with 67 additions and 109 deletions

View File

@@ -1,43 +0,0 @@
---
name: 🏁 Performance
type: Performance
description: An opportunity to improve application performance
labels: ["netbox", "type: performance", "status: needs triage"]
body:
- type: input
attributes:
label: NetBox Version
description: What version of NetBox are you currently running?
placeholder: v4.5.1
validations:
required: true
- type: dropdown
attributes:
label: Python Version
description: What version of Python are you currently running?
options:
- "3.12"
- "3.13"
- "3.14"
validations:
required: true
- type: checkboxes
attributes:
label: Area(s) of Concern
description: Which application interface(s) are affected?
options:
- label: User Interface
- label: REST API
- label: GraphQL API
- label: Python ORM
- label: Other
validations:
required: true
- type: textarea
attributes:
label: Details
description: >
Describe in detail the operations being performed and the indications of a performance issue.
Include any relevant testing parameters, benchmarks, and expected results.
validations:
required: true

View File

@@ -102,7 +102,10 @@ class GitBackend(DataBackend):
clone_args['pool_manager'] = ProxyPoolManager(self.socks_proxy)
if self.url_scheme in ('http', 'https'):
if self.params.get('username'):
# 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'):
clone_args.update(
{
"username": self.params.get('username'),

View File

@@ -0,0 +1,59 @@
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)

View File

@@ -43,7 +43,7 @@ IMAGEATTACHMENT_IMAGE = """
<a href="{{ record.image.url }}" target="_blank" class="image-preview" data-bs-placement="top">
<i class="mdi mdi-image"></i></a>
{% endif %}
<a href="{{ record.get_absolute_url }}">{{ record.filename|truncate_middle:16 }}</a>
<a href="{{ record.get_absolute_url }}">{{ record }}</a>
"""
NOTIFICATION_ICON = """

View File

@@ -6,7 +6,7 @@ from django.core.files.uploadedfile import SimpleUploadedFile
from django.forms import ValidationError
from django.test import tag, TestCase
from core.models import AutoSyncRecord, DataSource, ObjectType
from core.models import DataSource, ObjectType
from dcim.models import Device, DeviceRole, DeviceType, Location, Manufacturer, Platform, Region, Site, SiteGroup
from extras.models import ConfigContext, ConfigContextProfile, ConfigTemplate, ImageAttachment, Tag, TaggedItem
from tenancy.models import Tenant, TenantGroup
@@ -754,53 +754,3 @@ class ConfigTemplateTest(TestCase):
@tag('regression')
def test_config_template_with_data_source_nested_templates(self):
self.assertEqual(self.BASE_TEMPLATE, self.main_config_template.render({}))
@tag('regression')
def test_autosyncrecord_cleanup_on_detach(self):
"""Test that AutoSyncRecord is deleted when detaching from DataSource."""
with tempfile.TemporaryDirectory() as temp_dir:
templates_dir = Path(temp_dir) / "templates"
templates_dir.mkdir(parents=True, exist_ok=True)
self._create_template_file(templates_dir, 'test.j2', 'Test content')
data_source = DataSource(
name="Test DataSource for Detach",
type="local",
source_url=str(templates_dir),
)
data_source.save()
data_source.sync()
data_file = data_source.datafiles.filter(path__endswith='test.j2').first()
# Create a ConfigTemplate with data_file and auto_sync_enabled
config_template = ConfigTemplate(
name="TestTemplateForDetach",
data_file=data_file,
auto_sync_enabled=True
)
config_template.clean()
config_template.save()
# Verify AutoSyncRecord was created
object_type = ObjectType.objects.get_for_model(ConfigTemplate)
autosync_records = AutoSyncRecord.objects.filter(
object_type=object_type,
object_id=config_template.pk
)
self.assertEqual(autosync_records.count(), 1, "AutoSyncRecord should be created")
# Detach from DataSource
config_template.data_file = None
config_template.data_source = None
config_template.auto_sync_enabled = False
config_template.clean()
config_template.save()
# Verify AutoSyncRecord was deleted
autosync_records = AutoSyncRecord.objects.filter(
object_type=object_type,
object_id=config_template.pk
)
self.assertEqual(autosync_records.count(), 0, "AutoSyncRecord should be deleted after detaching")

View File

@@ -569,6 +569,7 @@ class SyncedDataMixin(models.Model):
)
else:
AutoSyncRecord.objects.filter(
datafile=self.data_file,
object_type=object_type,
object_id=self.pk
).delete()
@@ -581,6 +582,7 @@ class SyncedDataMixin(models.Model):
# Delete AutoSyncRecord
object_type = ObjectType.objects.get_for_model(self)
AutoSyncRecord.objects.filter(
datafile=self.data_file,
object_type=object_type,
object_id=self.pk
).delete()

View File

@@ -252,16 +252,3 @@ def isodatetime(value, spec='seconds'):
else:
return ''
return mark_safe(f'<span title="{naturaltime(value)}">{text}</span>')
@register.filter
def truncate_middle(value, length):
if len(value) <= length:
return value
# Calculate split points for the two parts
half_len = (length - 1) // 2 # 1 for the ellipsis
first_part = value[:half_len]
second_part = value[len(value) - (length - 1 - half_len):]
return mark_safe(f"{first_part}&hellip;{second_part}")