Coverage for extras/scripts.py: 29%

1import inspect 

2import logging 

3import os 

4import re 

5 

6from django import forms 

7from django.core.exceptions import ValidationError 

8from django.core.files.storage import storages 

9from django.core.validators import RegexValidator 

10from django.utils import timezone 

11from django.utils.functional import classproperty 

12from django.utils.translation import gettext as _ 

13from rq.exceptions import TimeoutFormatError 

14from rq.utils import parse_timeout 

15 

16from core.choices import JobNotificationChoices 

17from extras.choices import LogLevelChoices 

18from extras.constants import SCRIPT_MODULE_NAME_PREFIX 

19from extras.models import ScriptModule 

20from ipam.formfields import IPAddressFormField, IPNetworkFormField 

21from ipam.validators import MaxPrefixLengthValidator, MinPrefixLengthValidator, prefix_validator 

22from utilities.forms import add_blank_choice 

23from utilities.forms.fields import DynamicModelChoiceField, DynamicModelMultipleChoiceField 

24from utilities.forms.widgets import DatePicker, DateTimePicker 

25 

26from .forms import ScriptForm 

27 

28__all__ = ( 

29 'BaseScript', 

30 'BooleanVar', 

31 'ChoiceVar', 

32 'DateTimeVar', 

33 'DateVar', 

34 'DecimalVar', 

35 'FileVar', 

36 'IPAddressVar', 

37 'IPAddressWithMaskVar', 

38 'IPNetworkVar', 

39 'IntegerVar', 

40 'MultiChoiceVar', 

41 'MultiObjectVar', 

42 'ObjectVar', 

43 'Script', 

44 'StringVar', 

45 'TextVar', 

46 'get_module_and_script', 

47) 

48 

49# Internal ScriptForm fields used to carry execution parameters (see ScriptForm in 

50# extras/forms/scripts.py). These are validated/sourced separately from the script's own 

51# declared variables and must never be treated as script data or surfaced as script errors. 

52EXEC_PARAM_FIELDS = ('_commit', '_schedule_at', '_interval', '_notifications') 

53 

54# Sentinel distinguishing "argument not supplied" from an explicit None in validate_meta(). 

55_UNSET = object() 

56 

57 

58# 

59# Script variables 

60# 

61 

62class ScriptVariable: 

63 """ 

64 Base model for script variables 

65 """ 

66 form_field = forms.CharField 

67 

68 def __init__(self, label='', description='', default=None, required=True, widget=None): 

69 

70 # Initialize field attributes 

71 if not hasattr(self, 'field_attrs'): 

72 self.field_attrs = {} 

73 if label: 

74 self.field_attrs['label'] = label 

75 if description: 

76 self.field_attrs['help_text'] = description 

77 if default is not None: 

78 self.field_attrs['initial'] = default 

79 if widget: 

80 self.field_attrs['widget'] = widget 

81 self.field_attrs['required'] = required 

82 

83 def as_field(self): 

84 """ 

85 Render the variable as a Django form field. 

86 """ 

87 form_field = self.form_field(**self.field_attrs) 

88 if not isinstance(form_field.widget, forms.CheckboxInput): 

89 if form_field.widget.attrs and 'class' in form_field.widget.attrs.keys(): 

90 form_field.widget.attrs['class'] += ' form-control' 

91 else: 

92 form_field.widget.attrs['class'] = 'form-control' 

93 

94 return form_field 

95 

96 

97class StringVar(ScriptVariable): 

98 """ 

99 Character string representation. Can enforce minimum/maximum length and/or regex validation. 

100 """ 

101 def __init__(self, min_length=None, max_length=None, regex=None, *args, **kwargs): 

102 super().__init__(*args, **kwargs) 

103 

104 # Optional minimum/maximum lengths 

105 if min_length: 

106 self.field_attrs['min_length'] = min_length 

107 if max_length: 

108 self.field_attrs['max_length'] = max_length 

109 

110 # Optional regular expression validation 

111 if regex: 

112 self.field_attrs['validators'] = [ 

113 RegexValidator( 

114 regex=regex, 

115 message='Invalid value. Must match regex: {}'.format(regex), 

116 code='invalid' 

117 ) 

118 ] 

119 

120 

121class TextVar(ScriptVariable): 

122 """ 

123 Free-form text data. Renders as a <textarea>. 

124 """ 

125 form_field = forms.CharField 

126 

127 def __init__(self, *args, **kwargs): 

128 super().__init__(*args, **kwargs) 

129 

130 self.field_attrs['widget'] = forms.Textarea 

131 

132 

133class IntegerVar(ScriptVariable): 

134 """ 

135 Integer representation. Can enforce minimum/maximum values. 

136 """ 

137 form_field = forms.IntegerField 

138 

139 def __init__(self, min_value=None, max_value=None, *args, **kwargs): 

140 super().__init__(*args, **kwargs) 

141 

142 # Optional minimum/maximum values 

143 if min_value: 

144 self.field_attrs['min_value'] = min_value 

145 if max_value: 

146 self.field_attrs['max_value'] = max_value 

147 

148 

149class DecimalVar(ScriptVariable): 

150 """ 

151 Decimal representation. Can enforce minimum/maximum values, maximum digits and decimal places. 

152 """ 

153 form_field = forms.DecimalField 

154 

155 def __init__(self, min_value=None, max_value=None, max_digits=None, decimal_places=None, *args, **kwargs,): 

156 super().__init__(*args, **kwargs) 

157 

158 # Optional constraints 

159 if min_value: 

160 self.field_attrs["min_value"] = min_value 

161 if max_value: 

162 self.field_attrs["max_value"] = max_value 

163 if max_digits: 

164 self.field_attrs["max_digits"] = max_digits 

165 if decimal_places: 

166 self.field_attrs["decimal_places"] = decimal_places 

167 

168 

169class BooleanVar(ScriptVariable): 

170 """ 

171 Boolean representation (true/false). Renders as a checkbox. 

172 """ 

173 form_field = forms.BooleanField 

174 

175 def __init__(self, *args, **kwargs): 

176 super().__init__(*args, **kwargs) 

177 

178 # Boolean fields cannot be required 

179 self.field_attrs['required'] = False 

180 

181 

182class ChoiceVar(ScriptVariable): 

183 """ 

184 Select one of several predefined static choices, passed as a list of two-tuples. Example: 

185 

186 color = ChoiceVar( 

187 choices=( 

188 ('#ff0000', 'Red'), 

189 ('#00ff00', 'Green'), 

190 ('#0000ff', 'Blue') 

191 ) 

192 ) 

193 """ 

194 form_field = forms.ChoiceField 

195 

196 def __init__(self, choices, *args, **kwargs): 

197 super().__init__(*args, **kwargs) 

198 

199 # Set field choices, adding a blank choice to avoid forced selections 

200 self.field_attrs['choices'] = add_blank_choice(choices) 

201 

202 

203class DateVar(ScriptVariable): 

204 """ 

205 A date. 

206 """ 

207 form_field = forms.DateField 

208 

209 def __init__(self, *args, **kwargs): 

210 super().__init__(*args, **kwargs) 

211 self.form_field.widget = DatePicker() 

212 

213 

214class DateTimeVar(ScriptVariable): 

215 """ 

216 A date and a time. 

217 """ 

218 form_field = forms.DateTimeField 

219 

220 def __init__(self, *args, **kwargs): 

221 super().__init__(*args, **kwargs) 

222 self.form_field.widget = DateTimePicker() 

223 

224 

225class MultiChoiceVar(ScriptVariable): 

226 """ 

227 Like ChoiceVar, but allows for the selection of multiple choices. 

228 """ 

229 form_field = forms.MultipleChoiceField 

230 

231 def __init__(self, choices, *args, **kwargs): 

232 super().__init__(*args, **kwargs) 

233 

234 # Set field choices 

235 self.field_attrs['choices'] = choices 

236 

237 

238class ObjectVar(ScriptVariable): 

239 """ 

240 A single object within NetBox. 

241 

242 :param model: The NetBox model being referenced 

243 :param query_params: A dictionary of additional query parameters to attach when making REST API requests (optional) 

244 :param context: A custom dictionary mapping template context variables to fields, used when rendering <option> 

245 elements within the dropdown menu (optional) 

246 :param null_option: The label to use as a "null" selection option (optional) 

247 :param selector: Include an advanced object selection widget to assist the user in identifying the desired 

248 object (optional) 

249 :param quick_add: Include a widget to quickly create a new related object for assignment. (optional) 

250 """ 

251 form_field = DynamicModelChoiceField 

252 

253 def __init__(self, model, query_params=None, context=None, null_option=None, selector=False, quick_add=False, 

254 *args, **kwargs): 

255 super().__init__(*args, **kwargs) 

256 

257 self.field_attrs.update({ 

258 'queryset': model.objects.all(), 

259 'query_params': query_params, 

260 'context': context, 

261 'null_option': null_option, 

262 'selector': selector, 

263 'quick_add': quick_add, 

264 }) 

265 

266 

267class MultiObjectVar(ObjectVar): 

268 """ 

269 Like ObjectVar, but can represent one or more objects. 

270 """ 

271 form_field = DynamicModelMultipleChoiceField 

272 

273 

274class FileVar(ScriptVariable): 

275 """ 

276 An uploaded file. 

277 """ 

278 form_field = forms.FileField 

279 

280 

281class IPAddressVar(ScriptVariable): 

282 """ 

283 An IPv4 or IPv6 address without a mask. 

284 """ 

285 form_field = IPAddressFormField 

286 

287 

288class IPAddressWithMaskVar(ScriptVariable): 

289 """ 

290 An IPv4 or IPv6 address with a mask. 

291 """ 

292 form_field = IPNetworkFormField 

293 

294 

295class IPNetworkVar(ScriptVariable): 

296 """ 

297 An IPv4 or IPv6 prefix. 

298 """ 

299 form_field = IPNetworkFormField 

300 

301 def __init__(self, min_prefix_length=None, max_prefix_length=None, *args, **kwargs): 

302 super().__init__(*args, **kwargs) 

303 

304 # Set prefix validator and optional minimum/maximum prefix lengths 

305 self.field_attrs['validators'] = [prefix_validator] 

306 if min_prefix_length is not None: 

307 self.field_attrs['validators'].append( 

308 MinPrefixLengthValidator(min_prefix_length) 

309 ) 

310 if max_prefix_length is not None: 

311 self.field_attrs['validators'].append( 

312 MaxPrefixLengthValidator(max_prefix_length) 

313 ) 

314 

315 

316# 

317# Scripts 

318# 

319 

320class BaseScript: 

321 """ 

322 Base model for custom scripts. User classes should inherit from this model if they want to extend Script 

323 functionality for use in other subclasses. 

324 """ 

325 

326 # Prevent django from instantiating the class on all accesses 

327 do_not_call_in_templates = True 

328 

329 class Meta: 

330 pass 

331 

332 def __init__(self): 

333 self.messages = [] # Primary script log 

334 self.tests = {} # Mapping of logs for test methods 

335 self.output = '' 

336 self.failed = False 

337 self._current_test = None # Tracks the current test method being run (if any) 

338 

339 # Initiate the log 

340 self.logger = logging.getLogger(f"netbox.scripts.{self.full_name}") 

341 

342 # Declare the placeholder for the current request 

343 self.request = None 

344 

345 # Initiate the storage backend (local, S3, etc) as a class attr 

346 self.storage = storages.create_storage(storages.backends["scripts"]) 

347 

348 # Compile test methods and initialize results skeleton 

349 for method in dir(self): 

350 if method.startswith('test_') and callable(getattr(self, method)): 

351 self.tests[method] = { 

352 LogLevelChoices.LOG_SUCCESS: 0, 

353 LogLevelChoices.LOG_INFO: 0, 

354 LogLevelChoices.LOG_WARNING: 0, 

355 LogLevelChoices.LOG_FAILURE: 0, 

356 'log': [], 

357 } 

358 

359 def __str__(self): 

360 return self.name 

361 

362 @classproperty 

363 def module(self): 

364 # Strip the internal prefix applied when the module is loaded (see #22566) so that 

365 # user-facing names (full_name, logger namespaces) reflect the original script filename. 

366 name = self.__module__ 

367 if name.startswith(SCRIPT_MODULE_NAME_PREFIX): 

368 name = name[len(SCRIPT_MODULE_NAME_PREFIX):] 

369 return name 

370 

371 @classproperty 

372 def class_name(self): 

373 return self.__name__ 

374 

375 @classproperty 

376 def full_name(self): 

377 return f'{self.module}.{self.class_name}' 

378 

379 @classmethod 

380 def root_module(cls): 

381 return cls.module.split(".")[0] 

382 

383 # Author-defined attributes 

384 

385 @classproperty 

386 def name(self): 

387 return getattr(self.Meta, 'name', self.__name__) 

388 

389 @classproperty 

390 def description(self): 

391 return getattr(self.Meta, 'description', '') 

392 

393 @classproperty 

394 def field_order(self): 

395 return getattr(self.Meta, 'field_order', None) 

396 

397 @classproperty 

398 def fieldsets(self): 

399 return getattr(self.Meta, 'fieldsets', None) 

400 

401 @classproperty 

402 def commit_default(self): 

403 return getattr(self.Meta, 'commit_default', True) 

404 

405 @classproperty 

406 def job_timeout(self): 

407 return getattr(self.Meta, 'job_timeout', None) 

408 

409 @classproperty 

410 def scheduling_enabled(self): 

411 return getattr(self.Meta, 'scheduling_enabled', True) 

412 

413 @classproperty 

414 def notifications_default(self): 

415 return getattr(self.Meta, 'notifications_default', JobNotificationChoices.NOTIFICATION_ALWAYS) 

416 

417 @classmethod 

418 def validate_meta(cls, job_timeout=_UNSET, notifications=_UNSET): 

419 """ 

420 Validate the execution parameters used to run this script. Raises a ValidationError if any value is invalid, 

421 so that a misconfigured script surfaces an actionable error rather than an unhandled exception when the job is 

422 enqueued (see #22872). 

423 

424 The values actually enqueued are validated, not the raw Meta values: a caller may supply an explicit 

425 `job_timeout` or `notifications` (e.g. via the REST API), in which case that value is checked. When a caller 

426 omits a value, the corresponding Meta default is validated instead. Unset values fall back to valid defaults 

427 and are not rejected. 

428 """ 

429 errors = {} 

430 

431 job_timeout = cls.job_timeout if job_timeout is _UNSET else job_timeout 

432 if job_timeout is not None: 

433 # parse_timeout() is what RQ applies to the timeout downstream. It raises TimeoutFormatError for 

434 # malformed duration strings, but a job_timeout of an unexpected type (e.g. a list) instead raises 

435 # TypeError/ValueError/AssertionError from its internal int()/assert. Catch them all so any invalid value 

436 # surfaces as an actionable error rather than an unhandled 500. 

437 try: 

438 parsed_timeout = parse_timeout(job_timeout) 

439 except (TimeoutFormatError, TypeError, ValueError, AssertionError): 

440 parsed_timeout = None 

441 errors['job_timeout'] = _( 

442 "Invalid job_timeout value '{value}': must be an integer (seconds) or a duration string such as " 

443 "'1h' or '30m'." 

444 ).format(value=job_timeout) 

445 if parsed_timeout is not None and parsed_timeout <= 0: 

446 errors['job_timeout'] = _( 

447 "Invalid job_timeout value '{value}': must be a positive duration." 

448 ).format(value=job_timeout) 

449 

450 # A caller may pass notifications=None to mean "use the script's default"; treat that as unset. 

451 if notifications is _UNSET or notifications is None: 

452 notifications = cls.notifications_default 

453 if notifications not in JobNotificationChoices.values(): 

454 valid = ', '.join(JobNotificationChoices.values()) 

455 errors['notifications_default'] = _( 

456 "Invalid notifications value '{value}': must be one of {valid}." 

457 ).format(value=notifications, valid=valid) 

458 

459 if errors: 

460 raise ValidationError(errors) 

461 

462 @property 

463 def filename(self): 

464 return inspect.getfile(self.__class__) 

465 

466 def findsource(self, object): 

467 with self.storage.open(os.path.basename(self.filename), 'r') as f: 

468 data = f.read() 

469 

470 # Break the source code into lines 

471 lines = [line + '\n' for line in data.splitlines()] 

472 

473 # Find the class definition 

474 name = object.__name__ 

475 pat = re.compile(r'^(\s*)class\s*' + name + r'\b') 

476 # use the class definition with the least indentation 

477 candidates = [] 

478 for i in range(len(lines)): 

479 match = pat.match(lines[i]) 

480 if match: 

481 if lines[i][0] == 'c': 

482 return lines, i 

483 

484 candidates.append((match.group(1), i)) 

485 if not candidates: 

486 raise OSError('could not find class definition') 

487 

488 # Sort the candidates by whitespace, and by line number 

489 candidates.sort() 

490 return lines, candidates[0][1] 

491 

492 @property 

493 def source(self): 

494 # Can't use inspect.getsource() as it uses os to get the file 

495 # inspect uses ast, but that is overkill for this as we only do 

496 # classes. 

497 object = self.__class__ 

498 

499 try: 

500 lines, lnum = self.findsource(object) 

501 lines = inspect.getblock(lines[lnum:]) 

502 return ''.join(lines) 

503 except OSError: 

504 return '' 

505 

506 @classmethod 

507 def _get_vars(cls): 

508 vars = {} 

509 

510 # Iterate all base classes looking for ScriptVariables 

511 for base_class in inspect.getmro(cls): 

512 # When object is reached there's no reason to continue 

513 if base_class is object: 

514 break 

515 

516 for name, attr in base_class.__dict__.items(): 

517 if name not in vars and issubclass(attr.__class__, ScriptVariable): 

518 vars[name] = attr 

519 

520 # Order variables according to field_order 

521 if not cls.field_order: 

522 return vars 

523 ordered_vars = { 

524 field: vars.pop(field) for field in cls.field_order if field in vars 

525 } 

526 ordered_vars.update(vars) 

527 

528 return ordered_vars 

529 

530 def run(self, data, commit): 

531 """ 

532 Override this method with custom script logic. 

533 """ 

534 

535 # Backward compatibility for legacy Reports 

536 self.pre_run() 

537 self.run_tests() 

538 self.post_run() 

539 

540 def get_job_data(self): 

541 """ 

542 Return a dictionary of data to attach to the script's Job. 

543 """ 

544 return { 

545 'log': self.messages, 

546 'output': self.output, 

547 'tests': self.tests, 

548 } 

549 

550 # 

551 # Form rendering 

552 # 

553 

554 def get_fieldsets(self): 

555 fieldsets = [] 

556 

557 if self.fieldsets: 

558 fieldsets.extend(self.fieldsets) 

559 else: 

560 fields = list(name for name, __ in self._get_vars().items()) 

561 fieldsets.append((_('Script Data'), fields)) 

562 

563 # Append the default fieldset if defined in the Meta class 

564 if self.scheduling_enabled: 

565 exec_parameters = ('_schedule_at', '_interval', '_commit', '_notifications') 

566 else: 

567 exec_parameters = ('_commit', '_notifications') 

568 fieldsets.append((_('Script Execution Parameters'), exec_parameters)) 

569 

570 return fieldsets 

571 

572 def as_form(self, data=None, files=None, initial=None): 

573 """ 

574 Return a Django form suitable for populating the context data required to run this Script. 

575 """ 

576 # Create a dynamic ScriptForm subclass from script variables 

577 fields = { 

578 name: var.as_field() for name, var in self._get_vars().items() 

579 } 

580 FormClass = type('ScriptForm', (ScriptForm,), fields) 

581 

582 form = FormClass(data, files, initial=initial) 

583 

584 # Set initial "commit" checkbox state based on the script's Meta parameter 

585 form.fields['_commit'].initial = self.commit_default 

586 

587 # Set initial "notifications" selection based on the script's Meta parameter 

588 form.fields['_notifications'].initial = self.notifications_default 

589 

590 # Hide fields if scheduling has been disabled 

591 if not self.scheduling_enabled: 

592 form.fields['_schedule_at'].widget = forms.HiddenInput() 

593 form.fields['_interval'].widget = forms.HiddenInput() 

594 

595 return form 

596 

597 # 

598 # Logging 

599 # 

600 

601 def _log(self, message, obj=None, level=LogLevelChoices.LOG_INFO): 

602 """ 

603 Log a message. Do not call this method directly; use one of the log_* wrappers below. 

604 """ 

605 if level not in LogLevelChoices.values(): 

606 raise ValueError(f"Invalid logging level: {level}") 

607 

608 # A test method is currently active, so log the message using legacy Report logging 

609 if self._current_test: 

610 

611 # Increment the event counter for this level 

612 if level in self.tests[self._current_test]: 

613 self.tests[self._current_test][level] += 1 

614 

615 # Record message (if any) to the report log 

616 if message: 

617 # TODO: Use a dataclass for test method logs 

618 self.tests[self._current_test]['log'].append(( 

619 timezone.now().isoformat(), 

620 level, 

621 str(obj) if obj else None, 

622 obj.get_absolute_url() if hasattr(obj, 'get_absolute_url') else None, 

623 str(message), 

624 )) 

625 

626 elif message: 

627 

628 # Record to the script's log 

629 self.messages.append({ 

630 'time': timezone.now().isoformat(), 

631 'status': level, 

632 'message': str(message), 

633 'obj': str(obj) if obj else None, 

634 'url': obj.get_absolute_url() if hasattr(obj, 'get_absolute_url') else None, 

635 }) 

636 

637 # Record to the system log 

638 if obj: 

639 message = f"{obj}: {message}" 

640 self.logger.log(LogLevelChoices.SYSTEM_LEVELS[level], message) 

641 

642 def log_debug(self, message=None, obj=None): 

643 self._log(message, obj, level=LogLevelChoices.LOG_DEBUG) 

644 

645 def log_success(self, message=None, obj=None): 

646 self._log(message, obj, level=LogLevelChoices.LOG_SUCCESS) 

647 

648 def log_info(self, message=None, obj=None): 

649 self._log(message, obj, level=LogLevelChoices.LOG_INFO) 

650 

651 def log_warning(self, message=None, obj=None): 

652 self._log(message, obj, level=LogLevelChoices.LOG_WARNING) 

653 

654 def log_failure(self, message=None, obj=None): 

655 self._log(message, obj, level=LogLevelChoices.LOG_FAILURE) 

656 self.failed = True 

657 

658 # 

659 # Legacy Report functionality 

660 # 

661 

662 def run_tests(self): 

663 """ 

664 Run the report and save its results. Each test method will be executed in order. 

665 """ 

666 self.logger.info("Running report") 

667 try: 

668 for test_name in self.tests: 

669 self._current_test = test_name 

670 test_method = getattr(self, test_name) 

671 test_method() 

672 self._current_test = None 

673 except Exception as e: 

674 self._current_test = None 

675 self.post_run() 

676 raise e 

677 

678 def pre_run(self): 

679 """ 

680 Legacy method for operations performed immediately prior to running a Report. 

681 """ 

682 pass 

683 

684 def post_run(self): 

685 """ 

686 Legacy method for operations performed immediately after running a Report. 

687 """ 

688 pass 

689 

690 

691class Script(BaseScript): 

692 """ 

693 Classes which inherit this model will appear in the list of available scripts. 

694 """ 

695 pass 

696 

697 

698# 

699# Functions 

700# 

701 

702 

703def is_variable(obj): 

704 """ 

705 Returns True if the object is a ScriptVariable. 

706 """ 

707 return isinstance(obj, ScriptVariable) 

708 

709 

710def get_module_and_script(module_name, script_name): 

711 module = ScriptModule.objects.get(file_path=f'{module_name}.py') 

712 script = module.scripts.get(name=script_name) 

713 return module, script 

714 

715 

716def prepare_script_form(script_instance, data, files=None): 

717 """ 

718 Return a bound ScriptForm for the given Script instance, back-filling the declared 

719 `default` of any variable omitted from `data`. 

720 

721 `data` is copied rather than coerced to a plain dict, so a QueryDict retains the 

722 multi-value semantics a MultiObjectVar's multi-select field depends on. 

723 """ 

724 data = data.copy() if data is not None else {} 

725 for name, var in script_instance._get_vars().items(): 

726 if name in data: 

727 continue 

728 if (initial := var.field_attrs.get('initial')) is None: 

729 continue 

730 if isinstance(initial, (list, tuple)) and hasattr(data, 'setlist'): 

731 # Assigning a list to a QueryDict stores it as a single nested value, which a 

732 # multi-select widget reads back as one bogus choice. Set the values individually 

733 # so a MultiChoiceVar/MultiObjectVar default binds as it does for a plain dict. 

734 data.setlist(name, list(initial)) 

735 else: 

736 data[name] = initial 

737 return script_instance.as_form(data=data, files=files)