Coding Standards Document - COS301-SE-2025/CRISP GitHub Wiki
Table of Contents
- Introduction
- General Principles
- Python & Django Coding Standards
- File Structure and Organization
- Documentation Standards
- Error Handling Standards
- Testing Standards
- Version Control Standards
- Code Review Guidelines
- Tools and Configuration
- Django-Specific Quality Assurance
1. Introduction
This document outlines the coding standards and conventions for the CRISP Anonymization System project. These standards ensure uniformity, clarity, flexibility, reliability, and efficiency across our codebase. All team members must adhere to these guidelines to maintain code quality and facilitate collaboration.
1.1 Purpose
- Ensure consistent code style across the project
- Improve code readability and maintainability
- Facilitate team collaboration and code reviews
- Reduce onboarding time for new team members
- Minimize bugs through standardized practices
2. General Principles
2.1 Code Quality Principles
- Readability: Code should be self-documenting and easy to understand
- Consistency: Follow established patterns throughout the project
- Simplicity: Prefer simple, clear solutions over complex ones
- Maintainability: Write code that can be easily modified and extended
- Performance: Consider efficiency without sacrificing readability
2.2 Design Principles
- Single Responsibility Principle: Each class/function should have one clear purpose
- Open/Closed Principle: Classes should be open for extension, closed for modification
- Dependency Inversion: Depend on abstractions, not concrete implementations
- Strategy Pattern: Use strategy pattern for different anonymization approaches
3. Python & Django Coding Standards
3.1 Django-Specific Style Guide
We follow PEP 8 combined with Django coding style guidelines:
3.1.1 Django Naming Conventions
# Models: Singular, PascalCase
class ThreatIndicator(models.Model):
pass
class Institution(models.Model):
pass
class TTPData(models.Model):
pass
# Views: Descriptive class names
class ThreatFeedAPIView(APIView):
pass
class ThreatFeedListView(ListView):
pass
# Services: Business domain naming
class IndicatorService:
def process_stix_indicators(self, data: Dict[str, Any]) -> List[ThreatIndicator]:
pass
class STIXTaxiiService:
def fetch_threat_intelligence(self, collection_id: str) -> Dict[str, Any]:
pass
# Repositories: Data access layer naming
class ThreatFeedRepository:
def find_by_institution(self, institution_id: int) -> QuerySet:
pass
class IndicatorRepository:
def bulk_create_indicators(self, indicators: List[Dict]) -> None:
pass
# Serializers: API serialization
class ThreatFeedSerializer(serializers.ModelSerializer):
class Meta:
model = ThreatFeed
fields = '__all__'
# Management Commands: Action-oriented
class Command(BaseCommand):
help = 'Process TAXII threat intelligence feeds'
def handle(self, *args, **options):
pass
3.1.2 Model Conventions
# Good: Proper Django model structure
class ThreatIndicator(models.Model):
"""Model representing a cybersecurity threat indicator."""
# Primary fields
ioc_value = models.CharField(max_length=255, db_index=True)
ioc_type = models.CharField(max_length=50, choices=IOC_TYPE_CHOICES)
confidence = models.IntegerField(validators=[MinValueValidator(0), MaxValueValidator(100)])
# Relationships
institution = models.ForeignKey('Institution', on_delete=models.CASCADE, related_name='indicators')
ttp_data = models.ManyToManyField('TTPData', blank=True, related_name='indicators')
# Metadata
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
is_active = models.BooleanField(default=True)
class Meta:
db_table = 'threat_indicators'
ordering = ['-created_at']
indexes = [
models.Index(fields=['ioc_type', 'confidence']),
models.Index(fields=['institution', 'created_at']),
]
def __str__(self) -> str:
return f"{self.ioc_type}: {self.ioc_value}"
def get_absolute_url(self) -> str:
return reverse('indicator-detail', kwargs={'pk': self.pk})
3.1.3 Service Layer Patterns
# Good: Service layer with dependency injection
class IndicatorService:
"""Service for managing threat indicators with business logic."""
def __init__(self,
indicator_repo: IndicatorRepository,
anonymization_context: AnonymizationContext):
self.indicator_repo = indicator_repo
self.anonymization_context = anonymization_context
def process_stix_bundle(self, stix_bundle: Dict[str, Any],
anonymization_level: AnonymizationLevel = AnonymizationLevel.MEDIUM) -> ProcessingResult:
"""
Process STIX bundle and create threat indicators.
Args:
stix_bundle: STIX 2.x bundle containing indicators
anonymization_level: Level of anonymization to apply
Returns:
ProcessingResult with success/failure statistics
Raises:
STIXProcessingError: When bundle processing fails
ValidationError: When indicator data is invalid
"""
try:
# Extract indicators from bundle
indicators = self._extract_indicators_from_bundle(stix_bundle)
# Apply anonymization if needed
if anonymization_level != AnonymizationLevel.NONE:
indicators = self._anonymize_indicators(indicators, anonymization_level)
# Validate and save indicators
validated_indicators = self._validate_indicators(indicators)
saved_count = self.indicator_repo.bulk_create_indicators(validated_indicators)
return ProcessingResult(
success=True,
processed_count=len(indicators),
saved_count=saved_count,
errors=[]
)
except Exception as e:
logger.error(f"Failed to process STIX bundle: {e}")
raise STIXProcessingError(f"Bundle processing failed: {e}")
3.1.4 Repository Pattern Implementation
# Good: Repository pattern for data access
class IndicatorRepository:
"""Repository for threat indicator data access."""
def __init__(self, model_class: Type[models.Model] = ThreatIndicator):
self.model = model_class
def find_by_institution(self, institution_id: int) -> QuerySet[ThreatIndicator]:
"""Find all indicators for a specific institution."""
return self.model.objects.filter(
institution_id=institution_id,
is_active=True
).select_related('institution').prefetch_related('ttp_data')
def find_by_ioc_type(self, ioc_type: str, limit: int = 100) -> QuerySet[ThreatIndicator]:
"""Find indicators by IOC type with pagination."""
return self.model.objects.filter(
ioc_type=ioc_type,
is_active=True
).order_by('-confidence', '-created_at')[:limit]
def bulk_create_indicators(self, indicators_data: List[Dict[str, Any]]) -> int:
"""Bulk create threat indicators with conflict resolution."""
try:
indicators = [
self.model(**data) for data in indicators_data
]
# Use bulk_create with ignore_conflicts for performance
created_indicators = self.model.objects.bulk_create(
indicators,
ignore_conflicts=True,
batch_size=1000
)
return len(created_indicators)
except IntegrityError as e:
logger.warning(f"Integrity error during bulk create: {e}")
# Fallback to individual creation with conflict handling
return self._create_indicators_individually(indicators_data)
3.2 Import Organization for Django
# Django project imports follow specific ordering:
# 1. Standard library imports
import json
import logging
from datetime import datetime, timezone
from typing import Dict, List, Optional, Type, Union, Any
# 2. Third-party imports
import requests
from celery import shared_task
from django.core.exceptions import ValidationError
from django.db import models, transaction
from django.utils import timezone
from rest_framework import serializers, status
from rest_framework.decorators import api_view
from rest_framework.response import Response
# 3. Django project imports
from crisp.settings import TAXII_CONFIG
from core.models.indicator import ThreatIndicator
from core.models.institution import Institution
# 4. Local app imports (same app)
from .exceptions import STIXProcessingError
from .repositories.indicator_repository import IndicatorRepository
from .services.indicator_service import IndicatorService
# 5. Pattern imports (if using strategy/factory patterns)
from core.patterns.strategy.context import AnonymizationContext
from core.patterns.strategy.enums import AnonymizationLevel
from core.patterns.factory.stix_indicator_creator import STIXIndicatorCreator
3.3 Type Hints
All functions must include type hints:
# Good: Complete type hints
def execute_anonymization(self, data: str, data_type: DataType,
level: AnonymizationLevel) -> str:
pass
def bulk_anonymize(self, data_items: List[Tuple[str, DataType]],
level: AnonymizationLevel) -> List[str]:
pass
# Good: Complex type hints
def anonymize_stix_object(self, stix_data: Union[str, Dict[str, Any]],
level: AnonymizationLevel = AnonymizationLevel.MEDIUM) -> str:
pass
3.4 Error Handling Patterns
# Good: Specific exception handling with graceful degradation
def auto_detect_and_anonymize(self, data: str, level: AnonymizationLevel) -> str:
try:
data_type = self._detect_data_type(data)
return self.execute_anonymization(data, data_type, level)
except Exception as e:
# Log error in production implementation
return f"anonymized-data-{hash(data) % 10000}"
# Good: Input validation with clear error messages
def validate(self, data: str) -> bool:
try:
if '%' in data and ':' in data:
ip_part, _ = data.split('%', 1)
ipaddress.ip_address(ip_part)
else:
ipaddress.ip_address(data)
return True
except ValueError:
return False
4. File Structure and Organization
4.1 Repository Structure
CRISP/
├── core/ # Main Django application
│ ├── config/
│ │ └── taxii_sources.py # TAXII source configuration
│ ├── management/
│ │ └── commands/
│ │ ├── taxii_operations.py # TAXII management commands
│ │ └── test_taxii.py # Test TAXII command
│ ├── migrations/
│ │ └── 0001_initial.py # Database migrations
│ ├── models/
│ │ ├── __init__.py
│ │ ├── indicator.py # Threat indicator models
│ │ ├── institution.py # Institution models
│ │ └── ttp_data.py # TTP (Tactics, Techniques, Procedures) models
│ ├── parsers/
│ │ └── stix1_parser.py # STIX 1.x XML parser
│ ├── patterns/
│ │ ├── decorator/
│ │ │ ├── stix_decorator.py # STIX object decorators
│ │ │ └── stix_object_component.py
│ │ ├── factory/
│ │ │ ├── stix_indicator_creator.py # STIX indicator factory
│ │ │ ├── stix_object_creator.py # Base STIX object factory
│ │ │ └── stix_ttp_creator.py # STIX TTP factory
│ │ ├── observer/
│ │ │ └── threat_feed.py # Observer pattern for threat feeds
│ │ └── strategy/
│ │ ├── context.py # Strategy pattern context
│ │ ├── demo.py # Demo strategies
│ │ ├── enums.py # Strategy enumerations
│ │ ├── exceptions.py # Strategy exceptions
│ │ ├── strategies.py # Strategy implementations
│ │ └── utils.py # Strategy utilities
│ ├── repositories/
│ │ ├── indicator_repository.py # Indicator data access layer
│ │ ├── threat_feed_repository.py # Threat feed data access layer
│ │ └── ttp_repository.py # TTP data access layer
│ ├── serializers/
│ │ └── threat_feed_serializer.py # DRF serializers for API
│ ├── services/
│ │ ├── indicator_service.py # Indicator business logic
│ │ ├── otx_taxii_service.py # OTX TAXII 1.x service
│ │ ├── stix_taxii_service.py # STIX TAXII 2.x service
│ │ └── ttp_service.py # TTP business logic
│ ├── tasks/
│ │ └── taxii_tasks.py # Celery background tasks
│ ├── tests/
│ │ ├── test_decorator.py # Decorator pattern tests
│ │ ├── test_end_to_end.py # End-to-end integration tests
│ │ ├── test_management_commands.py # Management command tests
│ │ ├── test_observer.py # Observer pattern tests
│ │ ├── test_repository.py # Repository tests
│ │ ├── test_stix1_parser.py # STIX 1.x parser tests
│ │ ├── test_stix_factory.py # STIX factory tests
│ │ ├── test_stix_mock_data.py # Mock data for tests
│ │ ├── test_taxii_integration.py # TAXII integration tests
│ │ └── test_taxii_service.py # TAXII service tests
│ └── views/
│ ├── api/
│ │ └── threat_feed_views.py # REST API endpoints
│ └── home.py # Home page views
├── crisp/ # Django project configuration
│ ├── __init__.py
│ ├── asgi.py # ASGI configuration
│ ├── celery.py # Celery configuration
│ ├── settings.py # Django settings
│ ├── test_settings.py # Test-specific settings
│ ├── urls.py # URL routing
│ └── wsgi.py # WSGI configuration
├── manage.py # Django management script
├── requirements.txt # Python dependencies
├── .gitignore # Git ignore rules
├── .github/ # GitHub workflows
│ └── workflows/
│ └── ci.yml
└── README.md # Project README
4.2 Django Architecture Principles
The CRISP system follows Django best practices with a clean architecture approach:
4.2.1 Layer Separation
- Models Layer (
models/
): Data models and database schema - Services Layer (
services/
): Business logic and external integrations - Repository Layer (
repositories/
): Data access abstraction - Views Layer (
views/
): HTTP request/response handling - Serializers (
serializers/
): Data serialization for APIs
4.2.2 Design Patterns Implementation
- Strategy Pattern (
patterns/strategy/
): Anonymization strategies - Factory Pattern (
patterns/factory/
): STIX object creation - Decorator Pattern (
patterns/decorator/
): STIX object enhancement - Observer Pattern (
patterns/observer/
): Threat feed monitoring - Repository Pattern (
repositories/
): Data access layer
4.2.3 Django App Structure
# Good: Proper Django app organization
core/
├── models/ # Database models
├── views/ # Request handlers
│ ├── api/ # REST API views
│ └── home.py # Web interface views
├── services/ # Business logic
├── repositories/ # Data access layer
├── serializers/ # API serialization
├── tests/ # Comprehensive test suite
└── management/ # Custom Django commands
└── commands/
4.3 File Naming Conventions
4.3.1 Django-Specific Naming
- Models: Singular nouns in snake_case:
indicator.py
,institution.py
,ttp_data.py
- Views: Descriptive names with
_views.py
suffix:threat_feed_views.py
- Services: Business domain with
_service.py
suffix:indicator_service.py
,stix_taxii_service.py
- Repositories: Data access with
_repository.py
suffix:threat_feed_repository.py
- Serializers: API serialization with
_serializer.py
suffix:threat_feed_serializer.py
- Tasks: Background jobs with
_tasks.py
suffix:taxii_tasks.py
- Management Commands: Action-oriented names:
taxii_operations.py
,test_taxii.py
4.3.2 Pattern Implementation Files
- Strategy Pattern:
strategies.py
,context.py
,enums.py
- Factory Pattern:
*_creator.py
pattern:stix_indicator_creator.py
- Decorator Pattern:
*_decorator.py
pattern:stix_decorator.py
- Observer Pattern: Domain-specific names:
threat_feed.py
4.3.3 Test File Naming
- Pattern:
test_[component].py
:test_repository.py
,test_stix_factory.py
- Integration tests:
test_[integration_type].py
:test_taxii_integration.py
,test_end_to_end.py
- Mock data files:
test_*_mock_data.py
:test_stix_mock_data.py
5. Documentation Standards
5.1 Docstring Format
We use Google-style docstrings for consistency:
def execute_anonymization(self, data: str, data_type: DataType,
level: AnonymizationLevel) -> str:
"""
Execute anonymization using the appropriate strategy.
Args:
data: The data to anonymize
data_type: The type of data
level: The anonymization level to apply
Returns:
The anonymized data
Raises:
ValueError: If no suitable strategy is found
DataValidationError: If the data doesn't match the expected format
Example:
>>> context = AnonymizationContext()
>>> result = context.execute_anonymization("192.168.1.1",
... DataType.IP_ADDRESS,
... AnonymizationLevel.MEDIUM)
>>> print(result) # "192.168.x.x"
"""
5.2 Class Documentation
class AnonymizationContext:
"""
Context class that uses different anonymization strategies.
Implements the Strategy pattern for flexible anonymization of various
data types including IP addresses, domains, emails, and URLs.
The context maintains consistent mappings for anonymized values to ensure
the same input always produces the same output within a session.
Attributes:
_strategies: Dictionary mapping data types to their strategies
_value_mappings: Consistent value mappings for anonymization
_id_mappings: STIX ID mappings for consistent anonymization
Example:
>>> context = AnonymizationContext()
>>> result = context.auto_detect_and_anonymize("192.168.1.1",
... AnonymizationLevel.HIGH)
"""
5.3 Inline Comments
# Good: Explain why, not what
# Use consistent anonymization to ensure same input produces same output
if original in self._value_mappings:
return self._value_mappings[original]
# Good: Explain complex logic
# Handle IPv6 addresses with zone identifiers (like fe80::1%eth0)
if '%' in data and ':' in data:
ip_part, zone_id = data.split('%', 1)
6. Error Handling Standards
6.1 Exception Hierarchy
# Base exception for all anonymization errors
class AnonymizationError(Exception):
"""Base exception for anonymization errors"""
pass
class InvalidDataTypeError(AnonymizationError):
"""Raised when an invalid data type is provided"""
pass
class StrategyNotFoundError(AnonymizationError):
"""Raised when no suitable strategy is found for a data type"""
pass
class InvalidAnonymizationLevelError(AnonymizationError):
"""Raised when an invalid anonymization level is provided"""
pass
class DataValidationError(AnonymizationError):
"""Raised when data doesn't match the expected format"""
pass
6.2 Error Handling Patterns
# Good: Graceful degradation with logging
def auto_detect_and_anonymize(self, data: str, level: AnonymizationLevel) -> str:
try:
data_type = self._detect_data_type(data)
return self.execute_anonymization(data, data_type, level)
except Exception as e:
# In production, log the error for debugging
# logger.warning(f"Anonymization failed for {data}: {e}")
return f"anonymized-data-{hash(data) % 10000}"
# Good: Specific error handling
def validate(self, data: str) -> bool:
try:
ipaddress.ip_address(data)
return True
except ValueError:
return False # Invalid IP address format
6.3 Input Validation
# Always validate inputs at public method boundaries
def anonymize_stix_object(self, stix_data: Union[str, Dict[str, Any]],
level: AnonymizationLevel = AnonymizationLevel.MEDIUM) -> str:
"""Anonymize STIX object with validation."""
try:
# Parse and validate input
if isinstance(stix_data, str):
stix_obj = json.loads(stix_data)
else:
stix_obj = stix_data.copy()
self._validate_stix_input(stix_obj)
# ... rest of implementation
except Exception as e:
raise ValueError(f"Failed to anonymize STIX data: {e}")
7. Testing Standards
7.1 Django Test Structure
# Good: Django test organization
from django.test import TestCase, TransactionTestCase
from django.test.client import Client
from django.urls import reverse
from rest_framework.test import APITestCase, APIClient
from rest_framework import status
from unittest.mock import patch, MagicMock
class ThreatIndicatorModelTestCase(TestCase):
"""Test cases for ThreatIndicator model."""
def setUp(self):
"""Set up test fixtures before each test method."""
self.institution = Institution.objects.create(
name="Test Institution",
domain="test.edu"
)
self.indicator_data = {
'ioc_value': '192.168.1.100',
'ioc_type': 'ip',
'confidence': 85,
'institution': self.institution
}
def test_create_indicator_with_valid_data(self):
"""Test creating a threat indicator with valid data."""
# Arrange
expected_str = "ip: 192.168.1.100"
# Act
indicator = ThreatIndicator.objects.create(**self.indicator_data)
# Assert
self.assertEqual(str(indicator), expected_str)
self.assertTrue(indicator.is_active)
self.assertIsNotNone(indicator.created_at)
class IndicatorServiceTestCase(TestCase):
"""Test cases for IndicatorService business logic."""
def setUp(self):
"""Set up service dependencies."""
self.indicator_repo = IndicatorRepository()
self.anonymization_context = AnonymizationContext()
self.service = IndicatorService(
self.indicator_repo,
self.anonymization_context
)
self.institution = Institution.objects.create(
name="Test Institution",
domain="test.edu"
)
@patch('core.services.indicator_service.logger')
def test_process_stix_bundle_success(self, mock_logger):
"""Test successful STIX bundle processing."""
# Arrange
stix_bundle = {
"type": "bundle",
"id": "bundle--12345",
"objects": [
{
"type": "indicator",
"id": "indicator--67890",
"pattern": "[ipv4-addr:value = '192.168.1.100']",
"labels": ["malicious-activity"]
}
]
}
# Act
result = self.service.process_stix_bundle(
stix_bundle,
AnonymizationLevel.MEDIUM
)
# Assert
self.assertTrue(result.success)
self.assertEqual(result.processed_count, 1)
self.assertGreater(result.saved_count, 0)
class ThreatFeedAPITestCase(APITestCase):
"""Test cases for Threat Feed REST API."""
def setUp(self):
"""Set up API test client and test data."""
self.client = APIClient()
self.institution = Institution.objects.create(
name="Test Institution",
domain="test.edu"
)
# Create test indicators
ThreatIndicator.objects.create(
ioc_value='192.168.1.100',
ioc_type='ip',
confidence=85,
institution=self.institution
)
def test_get_threat_indicators_list(self):
"""Test retrieving list of threat indicators."""
# Arrange
url = reverse('threat-indicators-list')
# Act
response = self.client.get(url)
# Assert
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['ioc_value'], '192.168.1.100')
class TaxiiIntegrationTestCase(TransactionTestCase):
"""Integration tests for TAXII service interactions."""
def setUp(self):
"""Set up integration test environment."""
self.taxii_service = STIXTaxiiService()
self.test_collection_id = "test-collection-123"
@patch('requests.get')
def test_fetch_threat_intelligence_success(self, mock_get):
"""Test successful threat intelligence fetching from TAXII server."""
# Arrange
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"type": "bundle",
"objects": []
}
mock_get.return_value = mock_response
# Act
result = self.taxii_service.fetch_threat_intelligence(
self.test_collection_id
)
# Assert
self.assertIsInstance(result, dict)
self.assertEqual(result['type'], 'bundle')
mock_get.assert_called_once()
7.2 Test Coverage Requirements
- Minimum Coverage: 80% for all modules
- Critical Paths: 95% coverage for core business logic (services layer)
- Models: 90% coverage for model methods and properties
- APIs: 85% coverage for all API endpoints
- Integration: 75% coverage for external service integrations
- Pattern Implementations: 90% coverage for design pattern code
7.2.1 Django-Specific Testing Requirements
# Models testing
class ModelTestMixin:
"""Mixin for consistent model testing."""
def assert_model_str_representation(self, instance, expected_str):
"""Test model's string representation."""
self.assertEqual(str(instance), expected_str)
def assert_model_validation(self, model_class, invalid_data, expected_errors):
"""Test model validation."""
instance = model_class(**invalid_data)
with self.assertRaises(ValidationError) as context:
instance.full_clean()
for field, error in expected_errors.items():
self.assertIn(error, str(context.exception))
# API testing patterns
class APITestMixin:
"""Mixin for consistent API testing."""
def assert_api_response_structure(self, response, expected_fields):
"""Test API response structure."""
self.assertEqual(response.status_code, status.HTTP_200_OK)
for field in expected_fields:
self.assertIn(field, response.data)
def assert_api_error_response(self, response, expected_status, expected_message=None):
"""Test API error responses."""
self.assertEqual(response.status_code, expected_status)
if expected_message:
self.assertIn(expected_message, str(response.data))
7.3 Test Naming Conventions
# Django model tests
def test_create_threat_indicator_with_valid_data_succeeds(self):
pass
def test_threat_indicator_str_representation_returns_formatted_string(self):
pass
def test_threat_indicator_validation_with_invalid_confidence_raises_error(self):
pass
# Service layer tests
def test_indicator_service_process_stix_bundle_with_valid_data_creates_indicators(self):
pass
def test_indicator_service_anonymize_indicators_at_high_level_masks_sensitive_data(self):
pass
def test_indicator_service_bulk_create_with_duplicates_handles_conflicts_gracefully(self):
pass
# API endpoint tests
def test_get_threat_indicators_with_valid_institution_returns_filtered_results(self):
pass
def test_post_threat_feed_data_with_invalid_payload_returns_400_error(self):
pass
def test_threat_feed_api_authentication_without_token_returns_401_error(self):
pass
# Integration tests
def test_taxii_integration_fetch_from_otx_processes_stix_bundle_successfully(self):
pass
def test_end_to_end_stix_processing_with_anonymization_preserves_data_integrity(self):
pass
# Pattern implementation tests
def test_stix_factory_create_indicator_with_pattern_generates_correct_object(self):
pass
def test_anonymization_strategy_context_switches_between_strategies_correctly(self):
pass
8. Version Control Standards
8.1 Git Workflow
We follow GitFlow with the following branches:
main
: Production-ready codedevelop
: Integration branch for featuresfeature/*
: Individual feature developmenthotfix/*
: Critical bug fixesrelease/*
: Release preparation
8.2 Commit Message Format
type(scope): brief description
More detailed description if needed.
- Bullet points for multiple changes
- Each change on a new line
Closes #123
Types:
feat
: New featurefix
: Bug fixdocs
: Documentation changesstyle
: Code style changes (formatting, etc.)refactor
: Code refactoringtest
: Adding or updating testschore
: Maintenance tasks
Examples:
feat(strategies): add STIX 2.1 pattern anonymization support
- Implement pattern parsing for STIX indicators
- Add support for complex observables
- Update strategy to handle nested patterns
Closes #45
fix(context): handle IPv6 zone identifiers correctly
The context was failing to parse IPv6 addresses with zone
identifiers like fe80::1%eth0. Added proper handling.
Fixes #67
8.3 Branch Naming
feature/stix-bundle-anonymization
feature/improved-domain-detection
fix/ipv6-zone-id-parsing
hotfix/critical-security-vulnerability
release/v1.1.0
9. Code Review Guidelines
9.1 Review Checklist
Functionality:
- Code meets requirements
- Edge cases are handled
- Error conditions are properly managed
- Performance is acceptable
Code Quality:
- Follows coding standards
- Proper type hints are used
- Documentation is complete
- No code duplication
Testing:
- Unit tests are included
- Tests cover edge cases
- Integration tests pass
- Code coverage meets requirements
Security:
- No sensitive data exposed
- Input validation is proper
- Error messages don't leak information
9.2 Review Process
- Self-Review: Author reviews their own code first
- Automated Checks: CI/CD pipeline runs tests and linting
- Peer Review: At least one team member reviews
- Approval: Reviewer approves changes
- Merge: Changes are merged to target branch
10. Tools and Configuration
10.1 Django Development Tools
Linting and Formatting:
# Install Django development dependencies
pip install black flake8 mypy isort django-stubs
pip install pylint-django # Django-specific linting
# Format code (Django-aware)
black core/ crisp/
# Check style with Django extensions
flake8 core/ crisp/ --extend-ignore=E203,W503
pylint --load-plugins pylint_django core/
# Type checking with Django stubs
mypy core/ --django-settings-module=crisp.settings
# Sort imports (Django-aware)
isort core/ crisp/ --profile django
# Django-specific checks
python manage.py check
python manage.py check --deploy # Production readiness
Testing and Coverage:
# Run tests with coverage
coverage run --source='.' manage.py test
coverage report --show-missing
coverage html # Generate HTML coverage report
# Run specific test categories
python manage.py test core.tests.test_models
python manage.py test core.tests.test_services
python manage.py test core.tests.test_api
# Performance testing
python manage.py test --debug-mode
python manage.py test --parallel
Django Management Commands:
# Database operations
python manage.py makemigrations
python manage.py migrate
python manage.py dbshell
# Static analysis
python manage.py check --deploy
python manage.py validate_templates
# Custom CRISP commands
python manage.py taxii_operations --fetch-indicators
python manage.py test_taxii --collection-id=test-123
10.2 Configuration Files
Django-specific .flake8
:
[flake8]
max-line-length = 100
exclude =
.git,
__pycache__,
*/migrations/*,
*/venv/*,
build,
dist,
.env
ignore = E203,W503,E501
per-file-ignores =
*/settings.py:E501,F401
*/migrations/*.py:E501
*/tests/*.py:E501
Django pyproject.toml
:
[tool.black]
line-length = 100
target-version = ['py39']
include = '\.pyi?
extend-exclude = '''
/(
# directories
migrations
| venv
| .env
| build
| dist
)/
'''
[tool.isort]
profile = "django"
multi_line_output = 3
line_length = 100
known_django = "django"
known_first_party = "core,crisp"
sections = ["FUTURE","STDLIB","DJANGO","THIRDPARTY","FIRSTPARTY","LOCALFOLDER"]
[tool.mypy]
python_version = "3.9"
check_untyped_defs = true
ignore_missing_imports = true
warn_unused_ignores = true
warn_redundant_casts = true
warn_unused_configs = true
plugins = ["mypy_django_plugin.main"]
[tool.django-stubs]
django_settings_module = "crisp.settings"
[tool.coverage.run]
source = "."
omit = [
"*/migrations/*",
"*/venv/*",
"*/tests/*",
"manage.py",
"*/settings/*",
"*/wsgi.py",
"*/asgi.py"
]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise AssertionError",
"raise NotImplementedError"
]
Django settings.py
Configuration Standards:
# Good: Environment-based configuration
import os
from pathlib import Path
# Build paths
BASE_DIR = Path(__file__).resolve().parent.parent
# Security settings
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-key-change-in-production')
DEBUG = os.environ.get('DEBUG', 'False').lower() == 'true'
ALLOWED_HOSTS = os.environ.get('ALLOWED_HOSTS', 'localhost,127.0.0.1').split(',')
# Application definition
DJANGO_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
]
THIRD_PARTY_APPS = [
'rest_framework',
'corsheaders',
'celery',
]
LOCAL_APPS = [
'core',
]
INSTALLED_APPS = DJANGO_APPS + THIRD_PARTY_APPS + LOCAL_APPS
# Database configuration with environment variables
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': os.environ.get('DB_NAME', 'crisp_db'),
'USER': os.environ.get('DB_USER', 'crisp_user'),
'PASSWORD': os.environ.get('DB_PASSWORD', 'dev_password'),
'HOST': os.environ.get('DB_HOST', 'localhost'),
'PORT': os.environ.get('DB_PORT', '5432'),
}
}
# TAXII Configuration
TAXII_CONFIG = {
'OTX_API_KEY': os.environ.get('OTX_API_KEY'),
'TAXII_SERVER_URL': os.environ.get('TAXII_SERVER_URL', 'https://otx.alienvault.com/taxii/'),
'COLLECTION_TIMEOUT': int(os.environ.get('COLLECTION_TIMEOUT', '300')),
'MAX_INDICATORS_PER_BATCH': int(os.environ.get('MAX_INDICATORS_PER_BATCH', '1000')),
}
# Celery Configuration
CELERY_BROKER_URL = os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0')
CELERY_RESULT_BACKEND = os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/0')
CELERY_TIMEZONE = 'UTC'
CELERY_ENABLE_UTC = True
# REST Framework configuration
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.TokenAuthentication',
'rest_framework.authentication.SessionAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated',
],
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 50,
'DEFAULT_RENDERER_CLASSES': [
'rest_framework.renderers.JSONRenderer',
],
}
# Logging configuration
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
'style': '{',
},
},
'handlers': {
'file': {
'level': 'INFO',
'class': 'logging.FileHandler',
'filename': os.path.join(BASE_DIR, 'logs', 'django.log'),
'formatter': 'verbose',
},
'console': {
'level': 'DEBUG',
'class': 'logging.StreamHandler',
'formatter': 'verbose',
},
},
'loggers': {
'core': {
'handlers': ['file', 'console'],
'level': 'INFO',
'propagate': True,
},
'crisp': {
'handlers': ['file', 'console'],
'level': 'INFO',
'propagate': True,
},
},
}
10.3 Pre-commit Hooks for Django
.pre-commit-config.yaml
:
repos:
- repo: https://github.com/psf/black
rev: 22.10.0
hooks:
- id: black
language_version: python3.9
- repo: https://github.com/pycqa/flake8
rev: 5.0.4
hooks:
- id: flake8
additional_dependencies: [flake8-django]
- repo: https://github.com/pycqa/isort
rev: 5.10.1
hooks:
- id: isort
args: ["--profile", "django"]
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v0.991
hooks:
- id: mypy
additional_dependencies: [django-stubs]
- repo: local
hooks:
- id: django-check
name: Django Check
entry: python manage.py check
language: system
always_run: true
pass_filenames: false
- id: django-migrations-check
name: Django Migrations Check
entry: python manage.py makemigrations --check --dry-run
language: system
always_run: true
pass_filenames: false
10.4 IDE Configuration for Django
VS Code Settings (.vscode/settings.json
):
{
"python.defaultInterpreterPath": "./venv/bin/python",
"python.formatting.provider": "black",
"python.formatting.blackArgs": ["--line-length=100"],
"python.linting.enabled": true,
"python.linting.flake8Enabled": true,
"python.linting.mypyEnabled": true,
"python.linting.pylintEnabled": true,
"python.linting.pylintArgs": [
"--load-plugins=pylint_django",
"--django-settings-module=crisp.settings"
],
"python.sortImports.args": ["--profile", "django"],
"editor.formatOnSave": true,
"editor.codeActionsOnSave": {
"source.organizeImports": true
},
"python.testing.pytestEnabled": false,
"python.testing.unittestEnabled": true,
"python.testing.unittestArgs": [
"-v",
"-s",
".",
"-p",
"test*.py"
],
"files.associations": {
"*.html": "django-html"
},
"emmet.includeLanguages": {
"django-html": "html"
}
}
VS Code Extensions:
{
"recommendations": [
"ms-python.python",
"ms-python.flake8",
"ms-python.black-formatter",
"ms-python.isort",
"batisteo.vscode-django",
"ms-vscode.vscode-json",
"ms-python.mypy-type-checker"
]
}
10.5 Environment Configuration
.env.example
:
# Django Configuration
SECRET_KEY=your-secret-key-here
DEBUG=True
ALLOWED_HOSTS=localhost,127.0.0.1
# Database Configuration
DB_NAME=crisp_db
DB_USER=crisp_user
DB_PASSWORD=your_db_password
DB_HOST=localhost
DB_PORT=5432
# TAXII Configuration
OTX_API_KEY=your-otx-api-key
TAXII_SERVER_URL=https://otx.alienvault.com/taxii/
COLLECTION_TIMEOUT=300
MAX_INDICATORS_PER_BATCH=1000
# Celery Configuration
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/0
# Email Configuration (for notifications)
EMAIL_HOST=smtp.gmail.com
EMAIL_PORT=587
EMAIL_USE_TLS=True
[email protected]
EMAIL_HOST_PASSWORD=your-email-password
requirements.txt
Organization:
# Django Core
Django==4.2.7
djangorestframework==3.14.0
django-cors-headers==4.3.1
# Database
psycopg2-binary==2.9.7
# Task Queue
celery==5.3.1
redis==4.6.0
# STIX/TAXII Processing
stix2==3.0.1
taxii2-client==2.3.0
requests==2.31.0
# Development Tools
black==22.10.0
flake8==5.0.4
mypy==0.991
django-stubs==1.16.0
isort==5.10.1
pylint-django==2.5.3
# Testing
coverage==6.5.0
factory-boy==3.2.1
responses==0.21.0
# Documentation
Sphinx==5.3.0
sphinx-rtd-theme==1.1.1
11. Django-Specific Quality Assurance
11.1 Security Standards
# Good: Secure Django practices
class ThreatIndicatorAPIView(APIView):
"""Secure API endpoint for threat indicators."""
permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]
authentication_classes = [TokenAuthentication]
throttle_classes = [UserRateThrottle, AnonRateThrottle]
def get_queryset(self):
"""Return filtered queryset based on user permissions."""
user = self.request.user
# Ensure users only see their institution's data
if user.is_superuser:
return ThreatIndicator.objects.all()
else:
return ThreatIndicator.objects.filter(
institution__users=user
).select_related('institution')
def post(self, request):
"""Create new threat indicator with validation."""
serializer = ThreatIndicatorSerializer(data=request.data)
if serializer.is_valid():
# Validate user can create indicators for this institution
institution = serializer.validated_data.get('institution')
if not request.user.has_perm('core.add_threatindicator', institution):
return Response(
{'error': 'Permission denied'},
status=status.HTTP_403_FORBIDDEN
)
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
11.2 Performance Standards
# Good: Performance-optimized Django code
class OptimizedIndicatorRepository:
"""Performance-optimized repository for threat indicators."""
def get_indicators_with_relationships(self, institution_id: int) -> QuerySet:
"""Get indicators with optimized database queries."""
return ThreatIndicator.objects.filter(
institution_id=institution_id
).select_related(
'institution' # Avoid N+1 queries
).prefetch_related(
'ttp_data' # Efficiently load many-to-many
).only(
'id', 'ioc_value', 'ioc_type', 'confidence',
'institution__name', 'created_at'
) # Only load needed fields
def bulk_update_confidence_scores(self, updates: List[Dict[str, Any]]) -> int:
"""Bulk update confidence scores efficiently."""
indicators_to_update = []
for update_data in updates:
indicator = ThreatIndicator(
id=update_data['id'],
confidence=update_data['confidence']
)
indicators_to_update.append(indicator)
# Use bulk_update for performance
return ThreatIndicator.objects.bulk_update(
indicators_to_update,
['confidence'],
batch_size=1000
)
Conclusion
These coding standards ensure our CRISP Cybersecurity Threat Intelligence system maintains high quality, consistency, and reliability across all Django components. The standards cover:
- Django-specific patterns for models, views, services, and repositories
- Design pattern implementations including Strategy, Factory, Decorator, and Observer patterns
- Comprehensive testing strategies for models, services, APIs, and integrations
- Security and performance best practices for threat intelligence processing
- Tool configurations optimized for Django development workflow
All team members are expected to follow these guidelines and contribute to their continuous improvement through code reviews and team discussions.
Document Version History:
- v1.0: Initial version with basic Python standards
- v2.0: Updated for Demo 2 with comprehensive Django standards, testing requirements, and design pattern implementations