security-review-construction

建筑系统安全审查技能

此技能可确保所有建筑软件系统遵循安全最佳实践，保护敏感的项目数据、财务信息和商业情报。

何时启用

• 构建 ERP/BIM 系统集成
• 创建建筑仪表板
• 处理成本／财务数据
• 构建文档管理系统
• 创建用于现场数据收集的 API
• 与外部平台集成（Procore、PlanGrid 等）
• 处理分包商／供应商数据
• 处理付款申请

建筑行业特定安全注意事项

1. 财务数据保护

# CRITICAL: Construction financial data security

# ❌ NEVER Do This
project_budget = 15000000  # Hardcoded in source
margin_percentage = 0.18   # Business-sensitive info in code

# ✅ ALWAYS Do This
import os
from cryptography.fernet import Fernet

# Load from secure configuration
project_config = load_secure_config(os.environ['PROJECT_CONFIG_PATH'])

# Encrypt sensitive data at rest
def encrypt_financial_data(data: dict) -> bytes:
    key = os.environ.get('ENCRYPTION_KEY')
    f = Fernet(key)
    return f.encrypt(json.dumps(data).encode())

财务数据检查清单

• 静态存储的成本估算已加密
• 日志中不暴露利润率／加价数据
• 付款信息已令牌化
• 历史定价受到保护，避免竞争对手获取
• 开标前确保投标金额安全

2. BIM/CAD 数据安全

# BIM data often contains proprietary design information

# ❌ NEVER store BIM directly in public cloud without encryption
s3.upload_file('model.ifc', bucket='public-bucket')

# ✅ ALWAYS encrypt and control access
def upload_bim_secure(file_path: str, project_id: str):
    # Encrypt file
    encrypted_path = encrypt_file(file_path)

    # Generate pre-signed URL with expiration
    presigned_url = s3.generate_presigned_url(
        'get_object',
        Params={
            'Bucket': 'secure-bim-bucket',
            'Key': f'{project_id}/{os.path.basename(file_path)}'
        },
        ExpiresIn=3600  # 1 hour expiration
    )

    # Log access
    audit_log.info(f"BIM access granted: {project_id}")

    return presigned_url

BIM/CAD 检查清单

• 静态存储的 IFC/RVT 文件已加密
• 访问记录已记录，用于审计追踪
• 下载链接设有时限
• 版本控制包含访问追踪
• 错误消息中不包含设计数据

3. 分包商／供应商数据

# Subcontractor data includes business-sensitive information

class SubcontractorDataHandler:
    """Secure handling of subcontractor data"""

    # Fields that require encryption
    SENSITIVE_FIELDS = [
        'insurance_policy_number',
        'bank_account',
        'tax_id',
        'bonding_capacity',
        'historical_pricing'
    ]

    def store_subcontractor(self, data: dict) -> str:
        # Encrypt sensitive fields
        for field in self.SENSITIVE_FIELDS:
            if field in data:
                data[field] = self.encrypt(data[field])

        # Store with audit trail
        sub_id = self.db.insert(data)
        self.audit.log(f"Subcontractor created: {sub_id}")

        return sub_id

    def get_subcontractor(self, sub_id: str, requester_id: str) -> dict:
        # Check authorization
        if not self.can_access(requester_id, sub_id):
            raise PermissionError("Unauthorized access to subcontractor data")

        # Log access
        self.audit.log(f"Subcontractor accessed: {sub_id} by {requester_id}")

        # Return with decrypted sensitive fields (only to authorized users)
        return self.decrypt_sensitive_fields(self.db.get(sub_id))

供应商数据检查清单

• 保险／担保信息已加密
• 银行信息受到保护（符合 PCI 要求）
• UI 中的税号已屏蔽（仅显示末 4 位）
• 定价历史访问受到控制
• 证书到期通知受到保护

4. 现场数据收集安全

# Mobile/field data collection must be secure

from datetime import datetime, timedelta
import hashlib

class FieldDataCollector:
    """Secure field data collection"""

    def validate_photo_submission(self, photo_data: dict) -> bool:
        # Verify GPS timestamp is recent (within 24 hours)
        photo_time = datetime.fromisoformat(photo_data['timestamp'])
        if datetime.now() - photo_time > timedelta(hours=24):
            raise ValueError("Photo timestamp too old - possible replay attack")

        # Verify file hash matches
        file_hash = hashlib.sha256(photo_data['content']).hexdigest()
        if file_hash != photo_data['declared_hash']:
            raise ValueError("File integrity check failed")

        # Validate GPS coordinates are within project boundary
        if not self.is_within_project_bounds(
            photo_data['lat'],
            photo_data['lon'],
            photo_data['project_id']
        ):
            self.audit.warn(f"Photo from outside project bounds: {photo_data}")

        return True

    def submit_daily_report(self, report: dict, user_id: str) -> str:
        # Verify user is assigned to project
        if not self.is_assigned_to_project(user_id, report['project_id']):
            raise PermissionError("User not assigned to this project")

        # Sign report with user credentials
        report['signature'] = self.sign_report(report, user_id)
        report['submitted_at'] = datetime.now().isoformat()

        return self.db.insert(report)

现场数据检查清单

• 已验证 GPS 数据是否合理
• 已验证照片时间戳
• 已检查文件完整性（哈希处理）
• 提交数据时要求用户进行身份验证
• 离线数据同步受到保护

5. CWICR 数据库安全

# CWICR contains proprietary cost data

class CWICRAccessControl:
    """Access control for CWICR database"""

    TIERS = {
        'basic': ['public_rates', 'standard_descriptions'],
        'professional': ['regional_rates', 'productivity_factors'],
        'enterprise': ['custom_rates', 'historical_data', 'analytics']
    }

    def search(self, query: str, user_id: str) -> list:
        # Get user tier
        tier = self.get_user_tier(user_id)

        # Limit results based on tier
        allowed_fields = self.TIERS[tier]

        # Execute search with field restrictions
        results = self.vector_search(
            query=query,
            fields=allowed_fields,
            limit=self.get_tier_limit(tier)
        )

        # Log search for analytics
        self.audit.log(f"CWICR search: {user_id}, query='{query[:50]}...'")

        return results

    def export_data(self, user_id: str, format: str) -> bytes:
        # Enterprise only
        if self.get_user_tier(user_id) != 'enterprise':
            raise PermissionError("Export requires enterprise tier")

        # Watermark exported data
        data = self.get_exportable_data(user_id)
        watermarked = self.add_watermark(data, user_id)

        return watermarked

CWICR 检查清单

• 已实施分级访问控制
• 按用户／层级实施 API 速率限制
• 数据导出内容带有水印
• 限制批量下载
• 监控竞争对手的访问

6. 集成安全（Procore、PlanGrid 等）

# Secure OAuth integration with construction platforms

class ConstructionPlatformIntegration:
    """Secure integration with external platforms"""

    def __init__(self, platform: str):
        self.platform = platform
        # Load credentials from secure vault
        self.credentials = self.vault.get(f'{platform}_oauth')

    def authenticate(self) -> str:
        # Use OAuth 2.0 with PKCE
        code_verifier = secrets.token_urlsafe(32)
        code_challenge = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode()).digest()
        ).decode().rstrip('=')

        # Never store tokens in code or logs
        token = self.oauth_flow(code_verifier, code_challenge)

        # Store token securely
        self.secure_token_store.set(
            key=f'{self.platform}_token',
            value=token,
            ttl=token['expires_in']
        )

        return token

    def sync_data(self, project_id: str) -> dict:
        # Validate project access before sync
        if not self.has_project_access(project_id):
            raise PermissionError(f"No access to project {project_id}")

        # Rate limit syncs
        self.rate_limiter.check(f'sync_{self.platform}')

        # Sync with retry and error handling
        try:
            data = self.api_client.get_project_data(project_id)
            self.validate_incoming_data(data)
            return data
        except APIError as e:
            # Log error without sensitive details
            self.logger.error(f"Sync failed for {project_id}: {type(e).__name__}")
            raise

集成检查清单

• 已实施带 PKCE 的 OAuth 2.0
• 令牌存储在安全保管库中（而非环境变量中）
• 自动刷新令牌
• 遵守 API 速率限制
• 已验证 Webhook 签名
• 已验证传入数据

7. 文档管理安全

# Construction documents often contain confidential information

class SecureDocumentManager:
    """Secure document handling for construction"""

    # Document classification levels
    CLASSIFICATIONS = {
        'public': [],
        'internal': ['daily_reports', 'schedules'],
        'confidential': ['contracts', 'bids', 'financials'],
        'restricted': ['legal', 'hr', 'insurance']
    }

    def upload_document(self, file: bytes, metadata: dict, user_id: str) -> str:
        # Scan for malware
        if not self.malware_scan(file):
            raise SecurityError("Malware detected in uploaded file")

        # Classify document
        classification = self.classify_document(metadata)

        # Check user can upload to this classification
        if not self.can_upload(user_id, classification):
            raise PermissionError(f"Cannot upload {classification} documents")

        # Encrypt based on classification
        if classification in ['confidential', 'restricted']:
            file = self.encrypt(file)

        # Store with audit trail
        doc_id = self.storage.put(file, metadata)
        self.audit.log(f"Document uploaded: {doc_id} by {user_id}")

        return doc_id

    def download_document(self, doc_id: str, user_id: str) -> bytes:
        # Check access
        doc = self.storage.get_metadata(doc_id)
        if not self.can_access(user_id, doc['classification']):
            raise PermissionError("Access denied")

        # Log download
        self.audit.log(f"Document downloaded: {doc_id} by {user_id}")

        # Return decrypted content
        return self.decrypt(self.storage.get(doc_id))

文档检查清单

• 上传时扫描恶意软件
• 文档分类系统
• 基于角色的访问控制
• 下载审计日志记录
• 对敏感文档进行加密
• 执行保留策略

建筑系统部署前安全检查清单

数据保护

• 静态存储和传输中的财务数据均已加密
• BIM/CAD 文件受到访问控制保护
• 分包商个人身份信息（PII）受到保护（符合 GDPR/CCPA 要求）
• CWICR 数据访问已进行适当分级
• 已启用备份加密

身份验证与授权

• 管理员用户启用多因素身份验证
• 已实施基于角色的访问控制
• 会话管理安全（超时、单设备）
• 定期轮换 API 密钥
• OAuth 集成使用 PKCE

审计与合规

• 已记录所有数据访问
• 日志防篡改（仅追加）
• 已记录保留策略
• 提供用于审计的数据导出功能
• 自动生成合规报告

集成安全

• 所有外部 API 均已进行身份验证
• 已验证 Webhook 签名
• 已验证所有输入数据
• 已实施速率限制
• 错误消息已清理

现场数据安全

• 移动应用使用证书固定
• 离线数据已加密
• 已验证 GPS／时间戳
• 已验证照片完整性
• 使用安全同步协议

资源

• OWASP Top 10
• NIST Cybersecurity Framework
• CIS Controls for Construction
• ISO 27001 for Construction

---

请牢记：建筑行业数据包含财务、法律和竞争信息。数据泄露可能导致投标失败、法律责任和声誉受损。安全不是可有可无的。