/****************************************************************************** * arch/x86/mm.c * * Copyright (c) 2002-2005 K A Fraser * Copyright (c) 2004 Christian Limpach * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ /* * A description of the x86 page table API: * * Domains trap to do_mmu_update with a list of update requests. * This is a list of (ptr, val) pairs, where the requested operation * is *ptr = val. * * Reference counting of pages: * ---------------------------- * Each page has two refcounts: tot_count and type_count. * * TOT_COUNT is the obvious reference count. It counts all uses of a * physical page frame by a domain, including uses as a page directory, * a page table, or simple mappings via a PTE. This count prevents a * domain from releasing a frame back to the free pool when it still holds * a reference to it. * * TYPE_COUNT is more subtle. A frame can be put to one of three * mutually-exclusive uses: it might be used as a page directory, or a * page table, or it may be mapped writable by the domain [of course, a * frame may not be used in any of these three ways!]. * So, type_count is a count of the number of times a frame is being * referred to in its current incarnation. Therefore, a page can only * change its type when its type count is zero. * * Pinning the page type: * ---------------------- * The type of a page can be pinned/unpinned with the commands * MMUEXT_[UN]PIN_L?_TABLE. Each page can be pinned exactly once (that is, * pinning is not reference counted, so it can't be nested). * This is useful to prevent a page's type count falling to zero, at which * point safety checks would need to be carried out next time the count * is increased again. * * A further note on writable page mappings: * ----------------------------------------- * For simplicity, the count of writable mappings for a page may not * correspond to reality. The 'writable count' is incremented for every * PTE which maps the page with the _PAGE_RW flag set. However, for * write access to be possible the page directory entry must also have * its _PAGE_RW bit set. We do not check this as it complicates the * reference counting considerably [consider the case of multiple * directory entries referencing a single page table, some with the RW * bit set, others not -- it starts getting a bit messy]. * In normal use, this simplification shouldn't be a problem. * However, the logic can be added if required. * * One more note on read-only page mappings: * ----------------------------------------- * We want domains to be able to map pages for read-only access. The * main reason is that page tables and directories should be readable * by a domain, but it would not be safe for them to be writable. * However, domains have free access to rings 1 & 2 of the Intel * privilege model. In terms of page protection, these are considered * to be part of 'supervisor mode'. The WP bit in CR0 controls whether * read-only restrictions are respected in supervisor mode -- if the * bit is clear then any mapped page is writable. * * We get round this by always setting the WP bit and disallowing * updates to it. This is very unlikely to cause a problem for guest * OS's, which will generally use the WP bit to simplify copy-on-write * implementation (in that case, OS wants a fault when it writes to * an application-supplied buffer). */ #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include /* * Mapping of first 2 or 4 megabytes of memory. This is mapped with 4kB * mappings to avoid type conflicts with fixed-range MTRRs covering the * lowest megabyte of physical memory. In any case the VGA hole should be * mapped with type UC. */ l1_pgentry_t __attribute__ ((__section__ (".bss.page_aligned"))) l1_identmap[L1_PAGETABLE_ENTRIES]; #define MEM_LOG(_f, _a...) gdprintk(XENLOG_WARNING , _f "\n" , ## _a) /* * PTE updates can be done with ordinary writes except: * 1. Debug builds get extra checking by using CMPXCHG[8B]. * 2. PAE builds perform an atomic 8-byte store with CMPXCHG8B. */ #if !defined(NDEBUG) || defined(__i386__) #define PTE_UPDATE_WITH_CMPXCHG #endif /* Used to defer flushing of memory structures. */ struct percpu_mm_info { #define DOP_FLUSH_TLB (1<<0) /* Flush the local TLB. */ #define DOP_FLUSH_ALL_TLBS (1<<1) /* Flush TLBs of all VCPUs of current dom. */ #define DOP_RELOAD_LDT (1<<2) /* Reload the LDT shadow mapping. */ unsigned int deferred_ops; /* If non-NULL, specifies a foreign subject domain for some operations. */ struct domain *foreign; }; static DEFINE_PER_CPU(struct percpu_mm_info, percpu_mm_info); /* * Returns the current foreign domain; defaults to the currently-executing * domain if a foreign override hasn't been specified. */ #define FOREIGNDOM (this_cpu(percpu_mm_info).foreign ?: current->domain) /* Private domain structs for DOMID_XEN and DOMID_IO. */ struct domain *dom_xen, *dom_io; /* Frame table and its size in pages. */ struct page_info *frame_table; unsigned long max_page; unsigned long total_pages; #define PAGE_CACHE_ATTRS (_PAGE_PAT|_PAGE_PCD|_PAGE_PWT) #define l1_disallow_mask(d) \ ((d != dom_io) && \ (rangeset_is_empty((d)->iomem_caps) && \ rangeset_is_empty((d)->arch.ioport_caps) && \ !has_arch_pdevs(d)) ? \ L1_DISALLOW_MASK : (L1_DISALLOW_MASK & ~PAGE_CACHE_ATTRS)) #ifdef CONFIG_COMPAT l2_pgentry_t *compat_idle_pg_table_l2 = NULL; #define l3_disallow_mask(d) (!is_pv_32on64_domain(d) ? \ L3_DISALLOW_MASK : \ COMPAT_L3_DISALLOW_MASK) #else #define l3_disallow_mask(d) L3_DISALLOW_MASK #endif static void queue_deferred_ops(struct domain *d, unsigned int ops) { ASSERT(d == current->domain); this_cpu(percpu_mm_info).deferred_ops |= ops; } void __init init_frametable(void) { unsigned long nr_pages, page_step, i, mfn; frame_table = (struct page_info *)FRAMETABLE_VIRT_START; nr_pages = PFN_UP(max_page * sizeof(*frame_table)); page_step = (1 << L2_PAGETABLE_SHIFT) >> PAGE_SHIFT; for ( i = 0; i < nr_pages; i += page_step ) { mfn = alloc_boot_pages(min(nr_pages - i, page_step), page_step); if ( mfn == 0 ) panic("Not enough memory for frame table\n"); map_pages_to_xen( FRAMETABLE_VIRT_START + (i << PAGE_SHIFT), mfn, page_step, PAGE_HYPERVISOR); } memset(frame_table, 0, nr_pages << PAGE_SHIFT); #if defined(__x86_64__) for ( i = 0; i < max_page; i ++ ) spin_lock_init(&frame_table[i].lock); #endif } void __init arch_init_memory(void) { extern void subarch_init_memory(void); unsigned long i, pfn, rstart_pfn, rend_pfn, iostart_pfn, ioend_pfn; /* * Initialise our DOMID_XEN domain. * Any Xen-heap pages that we will allow to be mapped will have * their domain field set to dom_xen. */ dom_xen = domain_create(DOMID_XEN, DOMCRF_dummy, 0); BUG_ON(dom_xen == NULL); /* * Initialise our DOMID_IO domain. * This domain owns I/O pages that are within the range of the page_info * array. Mappings occur at the priv of the caller. */ dom_io = domain_create(DOMID_IO, DOMCRF_dummy, 0); BUG_ON(dom_io == NULL); /* First 1MB of RAM is historically marked as I/O. */ for ( i = 0; i < 0x100; i++ ) share_xen_page_with_guest(mfn_to_page(i), dom_io, XENSHARE_writable); /* Any areas not specified as RAM by the e820 map are considered I/O. */ for ( i = 0, pfn = 0; pfn < max_page; i++ ) { while ( (i < e820.nr_map) && (e820.map[i].type != E820_RAM) && (e820.map[i].type != E820_UNUSABLE) ) i++; if ( i >= e820.nr_map ) { /* No more RAM regions: mark as I/O right to end of memory map. */ rstart_pfn = rend_pfn = max_page; } else { /* Mark as I/O just up as far as next RAM region. */ rstart_pfn = min_t(unsigned long, max_page, PFN_UP(e820.map[i].addr)); rend_pfn = max_t(unsigned long, rstart_pfn, PFN_DOWN(e820.map[i].addr + e820.map[i].size)); } /* * Make sure any Xen mappings of RAM holes above 1MB are blown away. * In particular this ensures that RAM holes are respected even in * the statically-initialised 1-16MB mapping area. */ iostart_pfn = max_t(unsigned long, pfn, 1UL << (20 - PAGE_SHIFT)); ioend_pfn = rstart_pfn; #if defined(CONFIG_X86_32) ioend_pfn = min_t(unsigned long, ioend_pfn, DIRECTMAP_MBYTES << (20 - PAGE_SHIFT)); #endif if ( iostart_pfn < ioend_pfn ) destroy_xen_mappings((unsigned long)mfn_to_virt(iostart_pfn), (unsigned long)mfn_to_virt(ioend_pfn)); /* Mark as I/O up to next RAM region. */ for ( ; pfn < rstart_pfn; pfn++ ) { BUG_ON(!mfn_valid(pfn)); share_xen_page_with_guest( mfn_to_page(pfn), dom_io, XENSHARE_writable); } /* Skip the RAM region. */ pfn = rend_pfn; } subarch_init_memory(); } int memory_is_conventional_ram(paddr_t p) { int i; for ( i = 0; i < e820.nr_map; i++ ) { if ( (e820.map[i].type == E820_RAM) && (e820.map[i].addr <= p) && (e820.map[i].size > p) ) return 1; } return 0; } unsigned long domain_get_maximum_gpfn(struct domain *d) { if ( is_hvm_domain(d) ) return d->arch.p2m->max_mapped_pfn; /* NB. PV guests specify nr_pfns rather than max_pfn so we adjust here. */ return arch_get_max_pfn(d) - 1; } void share_xen_page_with_guest( struct page_info *page, struct domain *d, int readonly) { if ( page_get_owner(page) == d ) return; set_gpfn_from_mfn(page_to_mfn(page), INVALID_M2P_ENTRY); spin_lock(&d->page_alloc_lock); /* The incremented type count pins as writable or read-only. */ page->u.inuse.type_info = (readonly ? PGT_none : PGT_writable_page); page->u.inuse.type_info |= PGT_validated | 1; page_set_owner(page, d); wmb(); /* install valid domain ptr before updating refcnt. */ ASSERT(page->count_info == 0); /* Only add to the allocation list if the domain isn't dying. */ if ( !d->is_dying ) { page->count_info |= PGC_allocated | 1; if ( unlikely(d->xenheap_pages++ == 0) ) get_knownalive_domain(d); list_add_tail(&page->list, &d->xenpage_list); } spin_unlock(&d->page_alloc_lock); } void share_xen_page_with_privileged_guests( struct page_info *page, int readonly) { share_xen_page_with_guest(page, dom_xen, readonly); } #if defined(__i386__) #ifdef NDEBUG /* Only PDPTs above 4GB boundary need to be shadowed in low memory. */ #define l3tab_needs_shadow(mfn) ((mfn) >= 0x100000) #else /* * In debug builds we shadow a selection of <4GB PDPTs to exercise code paths. * We cannot safely shadow the idle page table, nor shadow (v1) page tables * (detected by lack of an owning domain). As required for correctness, we * always shadow PDPTs above 4GB. */ #define l3tab_needs_shadow(mfn) \ (((((mfn) << PAGE_SHIFT) != __pa(idle_pg_table)) && \ (page_get_owner(mfn_to_page(mfn)) != NULL) && \ ((mfn) & 1)) || /* odd MFNs are shadowed */ \ ((mfn) >= 0x100000)) #endif static l1_pgentry_t *fix_pae_highmem_pl1e; /* Cache the address of PAE high-memory fixmap page tables. */ static int __init cache_pae_fixmap_address(void) { unsigned long fixmap_base = fix_to_virt(FIX_PAE_HIGHMEM_0); l2_pgentry_t *pl2e = virt_to_xen_l2e(fixmap_base); fix_pae_highmem_pl1e = l2e_to_l1e(*pl2e) + l1_table_offset(fixmap_base); return 0; } __initcall(cache_pae_fixmap_address); static DEFINE_PER_CPU(u32, make_cr3_timestamp); void make_cr3(struct vcpu *v, unsigned long mfn) /* Takes the MFN of a PAE l3 table, copies the contents to below 4GB if * necessary, and sets v->arch.cr3 to the value to load in CR3. */ { l3_pgentry_t *highmem_l3tab, *lowmem_l3tab; struct pae_l3_cache *cache = &v->arch.pae_l3_cache; unsigned int cpu = smp_processor_id(); /* Fast path: does this mfn need a shadow at all? */ if ( !l3tab_needs_shadow(mfn) ) { v->arch.cr3 = mfn << PAGE_SHIFT; /* Cache is no longer in use or valid */ cache->high_mfn = 0; return; } /* Caching logic is not interrupt safe. */ ASSERT(!in_irq()); /* Protects against pae_flush_pgd(). */ spin_lock(&cache->lock); cache->inuse_idx ^= 1; cache->high_mfn = mfn; /* Map the guest L3 table and copy to the chosen low-memory cache. */ l1e_write(fix_pae_highmem_pl1e-cpu, l1e_from_pfn(mfn, __PAGE_HYPERVISOR)); /* First check the previous high mapping can't be in the TLB. * (i.e. have we loaded CR3 since we last did this?) */ if ( unlikely(this_cpu(make_cr3_timestamp) == this_cpu(tlbflush_time)) ) flush_tlb_one_local(fix_to_virt(FIX_PAE_HIGHMEM_0 + cpu)); highmem_l3tab = (l3_pgentry_t *)fix_to_virt(FIX_PAE_HIGHMEM_0 + cpu); lowmem_l3tab = cache->table[cache->inuse_idx]; memcpy(lowmem_l3tab, highmem_l3tab, sizeof(cache->table[0])); l1e_write(fix_pae_highmem_pl1e-cpu, l1e_empty()); this_cpu(make_cr3_timestamp) = this_cpu(tlbflush_time); v->arch.cr3 = __pa(lowmem_l3tab); spin_unlock(&cache->lock); } #else /* !defined(__i386__) */ void make_cr3(struct vcpu *v, unsigned long mfn) { v->arch.cr3 = mfn << PAGE_SHIFT; } #endif /* !defined(__i386__) */ void write_ptbase(struct vcpu *v) { write_cr3(v->arch.cr3); } /* * Should be called after CR3 is updated. * * Uses values found in vcpu->arch.(guest_table and guest_table_user), and * for HVM guests, arch.monitor_table and hvm's guest CR3. * * Update ref counts to shadow tables appropriately. */ void update_cr3(struct vcpu *v) { unsigned long cr3_mfn=0; if ( paging_mode_enabled(v->domain) ) { paging_update_cr3(v); return; } #if CONFIG_PAGING_LEVELS == 4 if ( !(v->arch.flags & TF_kernel_mode) ) cr3_mfn = pagetable_get_pfn(v->arch.guest_table_user); else #endif cr3_mfn = pagetable_get_pfn(v->arch.guest_table); make_cr3(v, cr3_mfn); } static void invalidate_shadow_ldt(struct vcpu *v) { int i; unsigned long pfn; struct page_info *page; if ( v->arch.shadow_ldt_mapcnt == 0 ) return; v->arch.shadow_ldt_mapcnt = 0; for ( i = 16; i < 32; i++ ) { pfn = l1e_get_pfn(v->arch.perdomain_ptes[i]); if ( pfn == 0 ) continue; l1e_write(&v->arch.perdomain_ptes[i], l1e_empty()); page = mfn_to_page(pfn); ASSERT_PAGE_IS_TYPE(page, PGT_seg_desc_page); ASSERT_PAGE_IS_DOMAIN(page, v->domain); put_page_and_type(page); } /* Dispose of the (now possibly invalid) mappings from the TLB. */ if ( v == current ) queue_deferred_ops(v->domain, DOP_FLUSH_TLB | DOP_RELOAD_LDT); else flush_tlb_mask(v->domain->domain_dirty_cpumask); } static int alloc_segdesc_page(struct page_info *page) { struct desc_struct *descs; int i; descs = map_domain_page(page_to_mfn(page)); for ( i = 0; i < 512; i++ ) if ( unlikely(!check_descriptor(page_get_owner(page), &descs[i])) ) goto fail; unmap_domain_page(descs); return 0; fail: unmap_domain_page(descs); return -EINVAL; } /* Map shadow page at offset @off. */ int map_ldt_shadow_page(unsigned int off) { struct vcpu *v = current; struct domain *d = v->domain; unsigned long gmfn, mfn; l1_pgentry_t l1e, nl1e; unsigned long gva = v->arch.guest_context.ldt_base + (off << PAGE_SHIFT); int okay; BUG_ON(unlikely(in_irq())); guest_get_eff_kern_l1e(v, gva, &l1e); if ( unlikely(!(l1e_get_flags(l1e) & _PAGE_PRESENT)) ) return 0; gmfn = l1e_get_pfn(l1e); mfn = gmfn_to_mfn(d, gmfn); if ( unlikely(!mfn_valid(mfn)) ) return 0; okay = get_page_and_type(mfn_to_page(mfn), d, PGT_seg_desc_page); if ( unlikely(!okay) ) return 0; nl1e = l1e_from_pfn(mfn, l1e_get_flags(l1e) | _PAGE_RW); l1e_write(&v->arch.perdomain_ptes[off + 16], nl1e); v->arch.shadow_ldt_mapcnt++; return 1; } static int get_page_from_pagenr(unsigned long page_nr, struct domain *d) { struct page_info *page = mfn_to_page(page_nr); if ( unlikely(!mfn_valid(page_nr)) || unlikely(!get_page(page, d)) ) { MEM_LOG("Could not get page ref for pfn %lx", page_nr); return 0; } return 1; } static int get_page_and_type_from_pagenr(unsigned long page_nr, unsigned long type, struct domain *d, int preemptible) { struct page_info *page = mfn_to_page(page_nr); int rc; if ( unlikely(!get_page_from_pagenr(page_nr, d)) ) return -EINVAL; rc = (preemptible ? get_page_type_preemptible(page, type) : (get_page_type(page, type) ? 0 : -EINVAL)); if ( rc ) put_page(page); return rc; } /* * We allow root tables to map each other (a.k.a. linear page tables). It * needs some special care with reference counts and access permissions: * 1. The mapping entry must be read-only, or the guest may get write access * to its own PTEs. * 2. We must only bump the reference counts for an *already validated* * L2 table, or we can end up in a deadlock in get_page_type() by waiting * on a validation that is required to complete that validation. * 3. We only need to increment the reference counts for the mapped page * frame if it is mapped by a different root table. This is sufficient and * also necessary to allow validation of a root table mapping itself. */ #define define_get_linear_pagetable(level) \ static int \ get_##level##_linear_pagetable( \ level##_pgentry_t pde, unsigned long pde_pfn, struct domain *d) \ { \ unsigned long x, y; \ struct page_info *page; \ unsigned long pfn; \ \ if ( (level##e_get_flags(pde) & _PAGE_RW) ) \ { \ MEM_LOG("Attempt to create linear p.t. with write perms"); \ return 0; \ } \ \ if ( (pfn = level##e_get_pfn(pde)) != pde_pfn ) \ { \ /* Make sure the mapped frame belongs to the correct domain. */ \ if ( unlikely(!get_page_from_pagenr(pfn, d)) ) \ return 0; \ \ /* \ * Ensure that the mapped frame is an already-validated page table. \ * If so, atomically increment the count (checking for overflow). \ */ \ page = mfn_to_page(pfn); \ y = page->u.inuse.type_info; \ do { \ x = y; \ if ( unlikely((x & PGT_count_mask) == PGT_count_mask) || \ unlikely((x & (PGT_type_mask|PGT_validated)) != \ (PGT_##level##_page_table|PGT_validated)) ) \ { \ put_page(page); \ return 0; \ } \ } \ while ( (y = cmpxchg(&page->u.inuse.type_info, x, x + 1)) != x ); \ } \ \ return 1; \ } int is_iomem_page(unsigned long mfn) { return (!mfn_valid(mfn) || (page_get_owner(mfn_to_page(mfn)) == dom_io)); } int get_page_from_l1e( l1_pgentry_t l1e, struct domain *d) { unsigned long mfn = l1e_get_pfn(l1e); struct page_info *page = mfn_to_page(mfn); uint32_t l1f = l1e_get_flags(l1e); struct vcpu *curr = current; struct domain *owner; int okay; if ( !(l1f & _PAGE_PRESENT) ) return 1; if ( unlikely(l1f & l1_disallow_mask(d)) ) { MEM_LOG("Bad L1 flags %x", l1f & l1_disallow_mask(d)); return 0; } if ( is_iomem_page(mfn) ) { /* DOMID_IO reverts to caller for privilege checks. */ if ( d == dom_io ) d = curr->domain; if ( !iomem_access_permitted(d, mfn, mfn) ) { if ( mfn != (PADDR_MASK >> PAGE_SHIFT) ) /* INVALID_MFN? */ MEM_LOG("Non-privileged (%u) attempt to map I/O space %08lx", d->domain_id, mfn); return 0; } return 1; } /* * Let privileged domains transfer the right to map their target * domain's pages. This is used to allow stub-domain pvfb export to dom0, * until pvfb supports granted mappings. At that time this minor hack * can go away. */ owner = page_get_owner(page); if ( unlikely(d != owner) && (owner != NULL) && (d != curr->domain) && IS_PRIV_FOR(d, owner) ) d = owner; /* Foreign mappings into guests in shadow external mode don't * contribute to writeable mapping refcounts. (This allows the * qemu-dm helper process in dom0 to map the domain's memory without * messing up the count of "real" writable mappings.) */ okay = (((l1f & _PAGE_RW) && !(unlikely(paging_mode_external(d) && (d != curr->domain)))) ? get_page_and_type(page, d, PGT_writable_page) : get_page(page, d)); if ( !okay ) { MEM_LOG("Error getting mfn %lx (pfn %lx) from L1 entry %" PRIpte " for dom%d", mfn, get_gpfn_from_mfn(mfn), l1e_get_intpte(l1e), d->domain_id); } else if ( pte_flags_to_cacheattr(l1f) != ((page->count_info >> PGC_cacheattr_base) & 7) ) { uint32_t x, nx, y = page->count_info; uint32_t cacheattr = pte_flags_to_cacheattr(l1f); if ( is_xen_heap_page(page) ) { if ( (l1f & _PAGE_RW) && !(unlikely(paging_mode_external(d) && (d != curr->domain))) ) put_page_type(page); put_page(page); MEM_LOG("Attempt to change cache attributes of Xen heap page"); return 0; } while ( ((y >> PGC_cacheattr_base) & 7) != cacheattr ) { x = y; nx = (x & ~PGC_cacheattr_mask) | (cacheattr << PGC_cacheattr_base); y = cmpxchg(&page->count_info, x, nx); } #ifdef __x86_64__ map_pages_to_xen((unsigned long)mfn_to_virt(mfn), mfn, 1, PAGE_HYPERVISOR | cacheattr_to_pte_flags(cacheattr)); #endif } return okay; } /* NB. Virtual address 'l2e' maps to a machine address within frame 'pfn'. */ define_get_linear_pagetable(l2); static int get_page_from_l2e( l2_pgentry_t l2e, unsigned long pfn, struct domain *d) { int rc; if ( !(l2e_get_flags(l2e) & _PAGE_PRESENT) ) return 1; if ( unlikely((l2e_get_flags(l2e) & L2_DISALLOW_MASK)) ) { MEM_LOG("Bad L2 flags %x", l2e_get_flags(l2e) & L2_DISALLOW_MASK); return -EINVAL; } rc = get_page_and_type_from_pagenr( l2e_get_pfn(l2e), PGT_l1_page_table, d, 0); if ( unlikely(rc == -EINVAL) && get_l2_linear_pagetable(l2e, pfn, d) ) rc = 0; return rc; } define_get_linear_pagetable(l3); static int get_page_from_l3e( l3_pgentry_t l3e, unsigned long pfn, struct domain *d, int preemptible) { int rc; if ( !(l3e_get_flags(l3e) & _PAGE_PRESENT) ) return 1; if ( unlikely((l3e_get_flags(l3e) & l3_disallow_mask(d))) ) { MEM_LOG("Bad L3 flags %x", l3e_get_flags(l3e) & l3_disallow_mask(d)); return -EINVAL; } rc = get_page_and_type_from_pagenr( l3e_get_pfn(l3e), PGT_l2_page_table, d, preemptible); if ( unlikely(rc == -EINVAL) && get_l3_linear_pagetable(l3e, pfn, d) ) rc = 0; return rc; } #if CONFIG_PAGING_LEVELS >= 4 define_get_linear_pagetable(l4); static int get_page_from_l4e( l4_pgentry_t l4e, unsigned long pfn, struct domain *d, int preemptible) { int rc; if ( !(l4e_get_flags(l4e) & _PAGE_PRESENT) ) return 1; if ( unlikely((l4e_get_flags(l4e) & L4_DISALLOW_MASK)) ) { MEM_LOG("Bad L4 flags %x", l4e_get_flags(l4e) & L4_DISALLOW_MASK); return -EINVAL; } rc = get_page_and_type_from_pagenr( l4e_get_pfn(l4e), PGT_l3_page_table, d, preemptible); if ( unlikely(rc == -EINVAL) && get_l4_linear_pagetable(l4e, pfn, d) ) rc = 0; return rc; } #endif /* 4 level */ #ifdef __x86_64__ #ifdef USER_MAPPINGS_ARE_GLOBAL #define adjust_guest_l1e(pl1e, d) \ do { \ if ( likely(l1e_get_flags((pl1e)) & _PAGE_PRESENT) && \ likely(!is_pv_32on64_domain(d)) ) \ { \ /* _PAGE_GUEST_KERNEL page cannot have the Global bit set. */ \ if ( (l1e_get_flags((pl1e)) & (_PAGE_GUEST_KERNEL|_PAGE_GLOBAL)) \ == (_PAGE_GUEST_KERNEL|_PAGE_GLOBAL) ) \ MEM_LOG("Global bit is set to kernel page %lx", \ l1e_get_pfn((pl1e))); \ if ( !(l1e_get_flags((pl1e)) & _PAGE_USER) ) \ l1e_add_flags((pl1e), (_PAGE_GUEST_KERNEL|_PAGE_USER)); \ if ( !(l1e_get_flags((pl1e)) & _PAGE_GUEST_KERNEL) ) \ l1e_add_flags((pl1e), (_PAGE_GLOBAL|_PAGE_USER)); \ } \ } while ( 0 ) #else #define adjust_guest_l1e(pl1e, d) \ do { \ if ( likely(l1e_get_flags((pl1e)) & _PAGE_PRESENT) && \ likely(!is_pv_32on64_domain(d)) ) \ l1e_add_flags((pl1e), _PAGE_USER); \ } while ( 0 ) #endif #define adjust_guest_l2e(pl2e, d) \ do { \ if ( likely(l2e_get_flags((pl2e)) & _PAGE_PRESENT) && \ likely(!is_pv_32on64_domain(d)) ) \ l2e_add_flags((pl2e), _PAGE_USER); \ } while ( 0 ) #define adjust_guest_l3e(pl3e, d) \ do { \ if ( likely(l3e_get_flags((pl3e)) & _PAGE_PRESENT) ) \ l3e_add_flags((pl3e), likely(!is_pv_32on64_domain(d)) ? \ _PAGE_USER : \ _PAGE_USER|_PAGE_RW); \ } while ( 0 ) #define adjust_guest_l4e(pl4e, d) \ do { \ if ( likely(l4e_get_flags((pl4e)) & _PAGE_PRESENT) && \ likely(!is_pv_32on64_domain(d)) ) \ l4e_add_flags((pl4e), _PAGE_USER); \ } while ( 0 ) #else /* !defined(__x86_64__) */ #define adjust_guest_l1e(_p, _d) ((void)(_d)) #define adjust_guest_l2e(_p, _d) ((void)(_d)) #define adjust_guest_l3e(_p, _d) ((void)(_d)) #endif #ifdef CONFIG_COMPAT #define unadjust_guest_l3e(pl3e, d) \ do { \ if ( unlikely(is_pv_32on64_domain(d)) && \ likely(l3e_get_flags((pl3e)) & _PAGE_PRESENT) ) \ l3e_remove_flags((pl3e), _PAGE_USER|_PAGE_RW|_PAGE_ACCESSED); \ } while ( 0 ) #else #define unadjust_guest_l3e(_p, _d) ((void)(_d)) #endif void put_page_from_l1e(l1_pgentry_t l1e, struct domain *d) { unsigned long pfn = l1e_get_pfn(l1e); struct page_info *page; struct domain *e; struct vcpu *v; if ( !(l1e_get_flags(l1e) & _PAGE_PRESENT) || is_iomem_page(pfn) ) return; page = mfn_to_page(pfn); e = page_get_owner(page); /* * Check if this is a mapping that was established via a grant reference. * If it was then we should not be here: we require that such mappings are * explicitly destroyed via the grant-table interface. * * The upshot of this is that the guest can end up with active grants that * it cannot destroy (because it no longer has a PTE to present to the * grant-table interface). This can lead to subtle hard-to-catch bugs, * hence a special grant PTE flag can be enabled to catch the bug early. * * (Note that the undestroyable active grants are not a security hole in * Xen. All active grants can safely be cleaned up when the domain dies.) */ if ( (l1e_get_flags(l1e) & _PAGE_GNTTAB) && !d->is_shutting_down && !d->is_dying ) { MEM_LOG("Attempt to implicitly unmap a granted PTE %" PRIpte, l1e_get_intpte(l1e)); domain_crash(d); } /* Remember we didn't take a type-count of foreign writable mappings * to paging-external domains */ if ( (l1e_get_flags(l1e) & _PAGE_RW) && !(unlikely((e != d) && paging_mode_external(e))) ) { put_page_and_type(page); } else { /* We expect this is rare so we blow the entire shadow LDT. */ if ( unlikely(((page->u.inuse.type_info & PGT_type_mask) == PGT_seg_desc_page)) && unlikely(((page->u.inuse.type_info & PGT_count_mask) != 0)) && 
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from __future__ import absolute_import, division, print_function

import os

import pytest

from cryptography.exceptions import (
    AlreadyFinalized, InvalidSignature, _Reasons)
from cryptography.hazmat.primitives import hashes, interfaces
from cryptography.hazmat.primitives.asymmetric import dsa
from cryptography.utils import bit_length

from .fixtures_dsa import (
    DSA_KEY_1024, DSA_KEY_2048, DSA_KEY_3072
)
from ...utils import (
    der_encode_dsa_signature, load_fips_dsa_key_pair_vectors,
    load_fips_dsa_sig_vectors, load_vectors_from_file,
    raises_unsupported_algorithm
)


def _check_dsa_private_key(skey):
    assert skey
    assert skey.x
    assert skey.y
    assert skey.key_size

    skey_parameters = skey.parameters()
    assert skey_parameters
    assert skey_parameters.modulus
    assert skey_parameters.subgroup_order
    assert skey_parameters.generator
    assert skey_parameters.modulus == skey_parameters.p
    assert skey_parameters.subgroup_order == skey_parameters.q
    assert skey_parameters.generator == skey_parameters.g

    pkey = skey.public_key()
    assert pkey
    assert skey.y == pkey.y
    assert skey.key_size == pkey.key_size

    pkey_parameters = pkey.parameters()
    assert pkey_parameters
    assert pkey_parameters.modulus
    assert pkey_parameters.subgroup_order
    assert pkey_parameters.generator
    assert pkey_parameters.modulus == pkey_parameters.p
    assert pkey_parameters.subgroup_order == pkey_parameters.q
    assert pkey_parameters.generator == pkey_parameters.g

    assert skey_parameters.modulus == pkey_parameters.modulus
    assert skey_parameters.subgroup_order == pkey_parameters.subgroup_order
    assert skey_parameters.generator == pkey_parameters.generator


@pytest.mark.dsa
class TestDSA(object):
    def test_generate_dsa_parameters_class_method(self, backend):
        parameters = dsa.DSAParameters.generate(1024, backend)
        assert bit_length(parameters.p) == 1024

    def test_generate_dsa_parameters(self, backend):
        parameters = dsa.generate_parameters(1024, backend)
        assert isinstance(parameters, interfaces.DSAParameters)

    def test_generate_invalid_dsa_parameters(self, backend):
        with pytest.raises(ValueError):
            dsa.DSAParameters.generate(1, backend)

    @pytest.mark.parametrize(
        "vector",
        load_vectors_from_file(
            os.path.join(
                "asymmetric", "DSA", "FIPS_186-3", "KeyPair.rsp"),
            load_fips_dsa_key_pair_vectors
        )
    )
    def test_generate_dsa_keys(self, vector, backend):
        parameters = dsa.DSAParameterNumbers(
            p=vector['p'],
            q=vector['q'],
            g=vector['g']
        ).parameters(backend)
        skey = parameters.generate_private_key()
        if isinstance(skey, interfaces.DSAPrivateKeyWithNumbers):
            numbers = skey.private_numbers()
            skey_parameters = numbers.public_numbers.parameter_numbers
            pkey = skey.public_key()
            parameters = pkey.parameters()
            parameter_numbers = parameters.parameter_numbers()
            assert parameter_numbers.p == skey_parameters.p
            assert parameter_numbers.q == skey_parameters.q
            assert parameter_numbers.g == skey_parameters.g
            assert skey_parameters.p == vector['p']
            assert skey_parameters.q == vector['q']
            assert skey_parameters.g == vector['g']
            assert skey.key_size == bit_length(vector['p'])
            assert pkey.key_size == skey.key_size
            public_numbers = pkey.public_numbers()
            assert numbers.public_numbers.y == public_numbers.y
            assert numbers.public_numbers.y == pow(
                skey_parameters.g, numbers.x, skey_parameters.p
            )

    def test_generate_dsa_private_key_and_parameters(self, backend):
        skey = dsa.generate_private_key(1024, backend)
        assert skey
        if isinstance(skey, interfaces.DSAPrivateKeyWithNumbers):
            numbers = skey.private_numbers()
            skey_parameters = numbers.public_numbers.parameter_numbers
            assert numbers.public_numbers.y == pow(
                skey_parameters.g, numbers.x, skey_parameters.p
            )

    def test_invalid_parameters_argument_types(self):
        with pytest.raises(TypeError):
            dsa.DSAParameters(None, None, None)

    def test_invalid_private_key_argument_types(self):
        with pytest.raises(TypeError):
            dsa.DSAPrivateKey(None, None, None, None, None)

    def test_invalid_public_key_argument_types(self):
        with pytest.raises(TypeError):
            dsa.DSAPublicKey(None, None, None, None)

    def test_load_dsa_example_keys(self):
        parameters = dsa.DSAParameters(
            modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
            subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
            generator=DSA_KEY_1024.public_numbers.parameter_numbers.g
        )

        assert parameters
        assert parameters.modulus
        assert parameters.subgroup_order
        assert parameters.generator
        assert parameters.modulus == parameters.p
        assert parameters.subgroup_order == parameters.q
        assert parameters.generator == parameters.g

        pub_key = dsa.DSAPublicKey(
            modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
            subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
            generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
            y=DSA_KEY_1024.public_numbers.y
        )
        assert pub_key
        assert pub_key.key_size
        assert pub_key.y
        pub_key_parameters = pub_key.parameters()
        assert pub_key_parameters
        assert pub_key_parameters.modulus
        assert pub_key_parameters.subgroup_order
        assert pub_key_parameters.generator

        skey = dsa.DSAPrivateKey(
            modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
            subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
            generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
            y=DSA_KEY_1024.public_numbers.y,
            x=DSA_KEY_1024.x
        )
        assert skey
        _check_dsa_private_key(skey)
        skey_parameters = skey.parameters()
        assert skey_parameters
        assert skey_parameters.modulus
        assert skey_parameters.subgroup_order
        assert skey_parameters.generator

        pkey = dsa.DSAPublicKey(
            modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
            subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
            generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
            y=DSA_KEY_1024.public_numbers.y
        )
        assert pkey
        pkey_parameters = pkey.parameters()
        assert pkey_parameters
        assert pkey_parameters.modulus
        assert pkey_parameters.subgroup_order
        assert pkey_parameters.generator

        pkey2 = skey.public_key()
        assert pkey2
        pkey2_parameters = pkey.parameters()
        assert pkey2_parameters
        assert pkey2_parameters.modulus
        assert pkey2_parameters.subgroup_order
        assert pkey2_parameters.generator

        assert skey_parameters.modulus == pkey_parameters.modulus
        assert skey_parameters.subgroup_order == pkey_parameters.subgroup_order
        assert skey_parameters.generator == pkey_parameters.generator
        assert skey.y == pkey.y
        assert skey.key_size == pkey.key_size

        assert pkey_parameters.modulus == pkey2_parameters.modulus
        assert pkey_parameters.subgroup_order == \
            pkey2_parameters.subgroup_order
        assert pkey_parameters.generator == pkey2_parameters.generator
        assert pkey.y == pkey2.y
        assert pkey.key_size == pkey2.key_size

    def test_invalid_parameters_values(self):
        # Test a modulus < 1024 bits in length
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=2 ** 1000,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
            )

        # Test a modulus < 2048 bits in length
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=2 ** 2000,
                subgroup_order=DSA_KEY_2048.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g,
            )

        # Test a modulus < 3072 bits in length
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=2 ** 3000,
                subgroup_order=DSA_KEY_3072.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
            )

        # Test a modulus > 3072 bits in length
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=2 ** 3100,
                subgroup_order=DSA_KEY_3072.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
            )

        # Test a subgroup_order < 160 bits in length
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 150,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
            )

        # Test a subgroup_order < 256 bits in length
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_2048.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 250,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g
            )

        # Test a subgroup_order > 256 bits in length
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_3072.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 260,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
            )

        # Test a modulus, subgroup_order pair of (1024, 256) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_2048.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
            )

        # Test a modulus, subgroup_order pair of (2048, 160) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_2048.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g
            )

        # Test a modulus, subgroup_order pair of (3072, 160) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_3072.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
            )

        # Test a generator < 1
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=0
            )

        # Test a generator = 1
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=1
            )

        # Test a generator > modulus
        with pytest.raises(ValueError):
            dsa.DSAParameters(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=2 ** 1200
            )

    def test_invalid_dsa_private_key_arguments(self):
        # Test a modulus < 1024 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=2 ** 1000,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=DSA_KEY_1024.x,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a modulus < 2048 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=2 ** 2000,
                subgroup_order=DSA_KEY_2048.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g,
                x=DSA_KEY_2048.x,
                y=DSA_KEY_2048.public_numbers.y
            )

        # Test a modulus < 3072 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=2 ** 3000,
                subgroup_order=DSA_KEY_3072.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
                x=DSA_KEY_3072.x,
                y=DSA_KEY_3072.public_numbers.y
            )

        # Test a modulus > 3072 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=2 ** 3100,
                subgroup_order=DSA_KEY_3072.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
                x=DSA_KEY_3072.x,
                y=DSA_KEY_3072.public_numbers.y
            )

        # Test a subgroup_order < 160 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 150,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=DSA_KEY_1024.x,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a subgroup_order < 256 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_2048.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 250,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g,
                x=DSA_KEY_2048.x,
                y=DSA_KEY_2048.public_numbers.y
            )

        # Test a subgroup_order > 256 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_3072.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 260,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
                x=DSA_KEY_3072.x,
                y=DSA_KEY_3072.public_numbers.y
            )

        # Test a modulus, subgroup_order pair of (1024, 256) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_2048.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=DSA_KEY_1024.x,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a modulus, subgroup_order pair of (2048, 160) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_2048.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g,
                x=DSA_KEY_2048.x,
                y=DSA_KEY_2048.public_numbers.y
            )

        # Test a modulus, subgroup_order pair of (3072, 160) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_3072.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
                x=DSA_KEY_3072.x,
                y=DSA_KEY_3072.public_numbers.y
            )

        # Test a generator < 1
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=0,
                x=DSA_KEY_1024.x,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a generator = 1
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=1,
                x=DSA_KEY_1024.x,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a generator > modulus
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=2 ** 1200,
                x=DSA_KEY_1024.x,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test x = 0
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=0,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test x < 0
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=-2,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test x = subgroup_order
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=2 ** 159,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test x > subgroup_order
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=2 ** 200,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test y != (generator ** x) % modulus
        with pytest.raises(ValueError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=DSA_KEY_1024.x,
                y=2 ** 100
            )

        # Test a non-integer y value
        with pytest.raises(TypeError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=DSA_KEY_1024.x,
                y=None
            )

        # Test a non-integer x value
        with pytest.raises(TypeError):
            dsa.DSAPrivateKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                x=None,
                y=DSA_KEY_1024.public_numbers.y
            )

    def test_invalid_dsa_public_key_arguments(self):
        # Test a modulus < 1024 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=2 ** 1000,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a modulus < 2048 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=2 ** 2000,
                subgroup_order=DSA_KEY_2048.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g,
                y=DSA_KEY_2048.public_numbers.y
            )

        # Test a modulus < 3072 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=2 ** 3000,
                subgroup_order=DSA_KEY_3072.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
                y=DSA_KEY_3072.public_numbers.y
            )

        # Test a modulus > 3072 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=2 ** 3100,
                subgroup_order=DSA_KEY_3072.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
                y=DSA_KEY_3072.public_numbers.y
            )

        # Test a subgroup_order < 160 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 150,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a subgroup_order < 256 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_2048.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 250,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g,
                y=DSA_KEY_2048.public_numbers.y
            )

        # Test a subgroup_order > 256 bits in length
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_3072.public_numbers.parameter_numbers.p,
                subgroup_order=2 ** 260,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
                y=DSA_KEY_3072.public_numbers.y
            )

        # Test a modulus, subgroup_order pair of (1024, 256) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_2048.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a modulus, subgroup_order pair of (2048, 160) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_2048.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_2048.public_numbers.parameter_numbers.g,
                y=DSA_KEY_2048.public_numbers.y
            )

        # Test a modulus, subgroup_order pair of (3072, 160) bit lengths
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_3072.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_3072.public_numbers.parameter_numbers.g,
                y=DSA_KEY_3072.public_numbers.y
            )

        # Test a generator < 1
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=0,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a generator = 1
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=1,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a generator > modulus
        with pytest.raises(ValueError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=2 ** 1200,
                y=DSA_KEY_1024.public_numbers.y
            )

        # Test a non-integer y value
        with pytest.raises(TypeError):
            dsa.DSAPublicKey(
                modulus=DSA_KEY_1024.public_numbers.parameter_numbers.p,
                subgroup_order=DSA_KEY_1024.public_numbers.parameter_numbers.q,
                generator=DSA_KEY_1024.public_numbers.parameter_numbers.g,
                y=None
            )


@pytest.mark.dsa
class TestDSAVerification(object):
    _algorithms_dict = {
        'SHA1': hashes.SHA1,
        'SHA224': hashes.SHA224,
        'SHA256': hashes.SHA256,
        'SHA384': hashes.SHA384,
        'SHA512': hashes.SHA512
    }

    @pytest.mark.parametrize(
        "vector",
        load_vectors_from_file(
            os.path.join(
                "asymmetric", "DSA", "FIPS_186-3", "SigVer.rsp"),
            load_fips_dsa_sig_vectors
        )
    )
    def test_dsa_verification(self, vector, backend):
        digest_algorithm = vector['digest_algorithm'].replace("-", "")
        algorithm = self._algorithms_dict[digest_algorithm]
        if (
            not backend.dsa_parameters_supported(
                vector['p'], vector['q'], vector['g']
            ) or not backend.dsa_hash_supported(algorithm)
        ):
            pytest.skip(
                "{0} does not support the provided parameters".format(backend)
            )

        public_key = dsa.DSAPublicNumbers(
            parameter_numbers=dsa.DSAParameterNumbers(
                vector['p'], vector['q'], vector['g']
            ),
            y=vector['y']
        ).public_key(backend)
        sig = der_encode_dsa_signature(vector['r'], vector['s'])
        verifier = public_key.verifier(sig, algorithm())
        verifier.update(vector['msg'])
        if vector['result'] == "F":
            with pytest.raises(InvalidSignature):
                verifier.verify()
        else:
            verifier.verify()

    def test_dsa_verify_invalid_asn1(self, backend):
        parameters = pytest.deprecated_call(
            dsa.DSAParameters.generate,
            1024,
            backend
        )
        private_key = pytest.deprecated_call(
            dsa.DSAPrivateKey.generate,
            parameters,
            backend
        )
        public_key = pytest.deprecated_call(private_key.public_key)
        verifier = public_key.verifier(b'fakesig', hashes.SHA1(), backend)
        verifier.update(b'fakesig')
        with pytest.raises(InvalidSignature):
            verifier.verify()

    def test_use_after_finalize(self, backend):
        parameters = dsa.DSAParameters.generate(1024, backend)
        private_key = dsa.DSAPrivateKey.generate(parameters, backend)
        public_key = private_key.public_key()
        verifier = public_key.verifier(b'fakesig', hashes.SHA1(), backend)
        verifier.update(b'irrelevant')
        with pytest.raises(InvalidSignature):
            verifier.verify()
        with pytest.raises(AlreadyFinalized):
            verifier.verify()
        with pytest.raises(AlreadyFinalized):
            verifier.update(b"more data")

    def test_dsa_verifier_invalid_backend(self, backend):
        pretend_backend = object()
        params = dsa.DSAParameters.generate(1024, backend)
        private_key = dsa.DSAPrivateKey.generate(params, backend)
        public_key = private_key.public_key()

        with raises_unsupported_algorithm(
                _Reasons.BACKEND_MISSING_INTERFACE):
            public_key.verifier(b"sig", hashes.SHA1(), pretend_backend)


@pytest.mark.dsa
class TestDSASignature(object):
    _algorithms_dict = {
        'SHA1': hashes.SHA1,
        'SHA224': hashes.SHA224,
        'SHA256': hashes.SHA256,
        'SHA384': hashes.SHA384,
        'SHA512': hashes.SHA512}

    @pytest.mark.parametrize(
        "vector",
        load_vectors_from_file(
            os.path.join(
                "asymmetric", "DSA", "FIPS_186-3", "SigGen.txt"),
            load_fips_dsa_sig_vectors
        )
    )
    def test_dsa_signing(self, vector, backend):
        digest_algorithm = vector['digest_algorithm'].replace("-", "")
        algorithm = self._algorithms_dict[digest_algorithm]
        if (
            not backend.dsa_parameters_supported(
                vector['p'], vector['q'], vector['g']
            ) or not backend.dsa_hash_supported(algorithm)
        ):
            pytest.skip(
                "{0} does not support the provided parameters".format(backend)
            )

        private_key = dsa.DSAPrivateNumbers(
            public_numbers=dsa.DSAPublicNumbers(
                parameter_numbers=dsa.DSAParameterNumbers(
                    vector['p'], vector['q'], vector['g']
                ),
                y=vector['y']
            ),
            x=vector['x']
        ).private_key(backend)
        signer = private_key.signer(algorithm())
        signer.update(vector['msg'])
        signature = signer.finalize()
        assert signature

        public_key = private_key.public_key()
        verifier = public_key.verifier(signature, algorithm())
        verifier.update(vector['msg'])
        verifier.verify()

    def test_use_after_finalize(self, backend):
        parameters = dsa.DSAParameters.generate(1024, backend)
        private_key = dsa.DSAPrivateKey.generate(parameters, backend)
        signer = private_key.signer(hashes.SHA1(), backend)
        signer.update(b"data")
        signer.finalize()
        with pytest.raises(AlreadyFinalized):
            signer.finalize()
        with pytest.raises(AlreadyFinalized):
            signer.update(b"more data")

    def test_dsa_signer_invalid_backend(self, backend):
        pretend_backend = object()
        params = dsa.DSAParameters.generate(1024, backend)
        private_key = dsa.DSAPrivateKey.generate(params, backend)

        with raises_unsupported_algorithm(
                _Reasons.BACKEND_MISSING_INTERFACE):
            private_key.signer(hashes.SHA1(), pretend_backend)


def test_dsa_generate_invalid_backend():
    pretend_backend = object()

    with raises_unsupported_algorithm(
            _Reasons.BACKEND_MISSING_INTERFACE):
        dsa.DSAParameters.generate(1024, pretend_backend)

    pretend_parameters = object()
    with raises_unsupported_algorithm(
            _Reasons.BACKEND_MISSING_INTERFACE):
        dsa.DSAPrivateKey.generate(pretend_parameters, pretend_backend)


class TestDSANumbers(object):
    def test_dsa_parameter_numbers(self):
        parameter_numbers = dsa.DSAParameterNumbers(p=1, q=2, g=3)
        assert parameter_numbers.p == 1
        assert parameter_numbers.q == 2
        assert parameter_numbers.g == 3

    def test_dsa_parameter_numbers_invalid_types(self):
        with pytest.raises(TypeError):
            dsa.DSAParameterNumbers(p=None, q=2, g=3)

        with pytest.raises(TypeError):
            dsa.DSAParameterNumbers(p=1, q=None, g=3)

        with pytest.raises(TypeError):
            dsa.DSAParameterNumbers(p=1, q=2, g=None)

    def test_dsa_public_numbers(self):
        parameter_numbers = dsa.DSAParameterNumbers(p=1, q=2, g=3)
        public_numbers = dsa.DSAPublicNumbers(
            y=4,
            parameter_numbers=parameter_numbers
        )
        assert public_numbers.y == 4
        assert public_numbers.parameter_numbers == parameter_numbers

    def test_dsa_public_numbers_invalid_types(self):
        with pytest.raises(TypeError):
            dsa.DSAPublicNumbers(y=4, parameter_numbers=None)

        with pytest.raises(TypeError):
            parameter_numbers = dsa.DSAParameterNumbers(p=1, q=2, g=3)
            dsa.DSAPublicNumbers(y=None, parameter_numbers=parameter_numbers)

    def test_dsa_private_numbers(self):
        parameter_numbers = dsa.DSAParameterNumbers(p=1, q=2, g=3)
        public_numbers = dsa.DSAPublicNumbers(
            y=4,
            parameter_numbers=parameter_numbers
        )
        private_numbers = dsa.DSAPrivateNumbers(
            x=5,
            public_numbers=public_numbers
        )
        assert private_numbers.x == 5
        assert private_numbers.public_numbers == public_numbers

    def test_dsa_private_numbers_invalid_types(self):
        parameter_numbers = dsa.DSAParameterNumbers(p=1, q=2, g=3)
        public_numbers = dsa.DSAPublicNumbers(
            y=4,
            parameter_numbers=parameter_numbers
        )
        with pytest.raises(TypeError):
            dsa.DSAPrivateNumbers(x=4, public_numbers=None)

        with pytest.raises(TypeError):
            dsa.DSAPrivateNumbers(x=None, public_numbers=public_numbers)
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-23 05:59:56 +0000